<!doctype html>



  


<html class="theme-next muse use-motion">
<head>
  <meta charset="UTF-8"/>
<meta http-equiv="X-UA-Compatible" content="IE=edge,chrome=1" />
<meta name="viewport" content="width=device-width, initial-scale=1, maximum-scale=1"/>



<meta http-equiv="Cache-Control" content="no-transform" />
<meta http-equiv="Cache-Control" content="no-siteapp" />












  
  
  <link href="/vendors/fancybox/source/jquery.fancybox.css?v=2.1.5" rel="stylesheet" type="text/css" />




  
  
  
  

  
    
    
  

  

  

  

  

  
    
    
    <link href="//fonts.googleapis.com/css?family=Lato:300,300italic,400,400italic,700,700italic&subset=latin,latin-ext" rel="stylesheet" type="text/css">
  






<link href="/vendors/font-awesome/css/font-awesome.min.css?v=4.4.0" rel="stylesheet" type="text/css" />

<link href="/css/main.css?v=5.0.1" rel="stylesheet" type="text/css" />


  <meta name="keywords" content="zeromq,mq," />








  <link rel="shortcut icon" type="image/x-icon" href="/favicon.ico?v=5.0.1" />






<meta name="description" content="ZeroMQ的研究与学习


简介
工作模式
层级模型
实现原理
核心特点
与其他MQ的简单比较">
<meta property="og:type" content="article">
<meta property="og:title" content="从源码分析ZeroMQ">
<meta property="og:url" content="http://yoursite.com/2016/08/28/ZeroMQ/index.html">
<meta property="og:site_name" content="个人博客">
<meta property="og:description" content="ZeroMQ的研究与学习


简介
工作模式
层级模型
实现原理
核心特点
与其他MQ的简单比较">
<meta property="og:image" content="http://img0.imgtn.bdimg.com/it/u=2368283034,2384299468&fm=21&gp=0.jpg">
<meta property="og:image" content="http://pic004.cnblogs.com/news/201208/20120813_221017_1.jpg">
<meta property="og:image" content="http://pic004.cnblogs.com/news/201208/20120813_221027_1.jpg">
<meta property="og:image" content="https://github.com/imatix/zguide/raw/master/images/fig28.png">
<meta property="og:image" content="https://github.com/imatix/zguide/raw/master/images/fig26.png">
<meta property="og:image" content="http://pic004.cnblogs.com/news/201208/20120813_221017_2.jpg">
<meta property="og:image" content="http://img.blog.csdn.net/20160530140759747">
<meta property="og:image" content="http://img.blog.csdn.net/20160530185222726">
<meta property="og:image" content="http://img.blog.csdn.net/20131205133912640?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvZmpzbG92ZWpobA==/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/SouthEast">
<meta property="og:image" content="http://blog.chinaunix.net/attachment/201303/31/22312037_1364725293Ce0Y.png">
<meta property="og:image" content="http://blog.chinaunix.net/attachment/201303/31/22312037_1364725293Ce0Y.png">
<meta property="og:updated_time" content="2017-01-19T07:14:14.008Z">
<meta name="twitter:card" content="summary">
<meta name="twitter:title" content="从源码分析ZeroMQ">
<meta name="twitter:description" content="ZeroMQ的研究与学习


简介
工作模式
层级模型
实现原理
核心特点
与其他MQ的简单比较">
<meta name="twitter:image" content="http://img0.imgtn.bdimg.com/it/u=2368283034,2384299468&fm=21&gp=0.jpg">



<script type="text/javascript" id="hexo.configuration">
  var NexT = window.NexT || {};
  var CONFIG = {
    scheme: 'Muse',
    sidebar: {"position":"left","display":"post"},
    fancybox: true,
    motion: true,
    duoshuo: {
      userId: 0,
      author: '博主'
    }
  };
</script>




  <link rel="canonical" href="http://yoursite.com/2016/08/28/ZeroMQ/"/>

  <title> 从源码分析ZeroMQ | 个人博客 </title>
</head>

<body itemscope itemtype="http://schema.org/WebPage" lang="zh-Hans">

  










  
  
    
  

  <div class="container one-collumn sidebar-position-left page-post-detail ">
    <div class="headband"></div>

    <header id="header" class="header" itemscope itemtype="http://schema.org/WPHeader">
      <div class="header-inner"><div class="site-meta ">
  

  <div class="custom-logo-site-title">
    <a href="/"  class="brand" rel="start">
      <span class="logo-line-before"><i></i></span>
      <span class="site-title">个人博客</span>
      <span class="logo-line-after"><i></i></span>
    </a>
  </div>
  <p class="site-subtitle">随便写点什么</p>
</div>

<div class="site-nav-toggle">
  <button>
    <span class="btn-bar"></span>
    <span class="btn-bar"></span>
    <span class="btn-bar"></span>
  </button>
</div>

<nav class="site-nav">
  

  
    <ul id="menu" class="menu">
      
        
        <li class="menu-item menu-item-home">
          <a href="/" rel="section">
            
              <i class="menu-item-icon fa fa-fw fa-home"></i> <br />
            
            首页
          </a>
        </li>
      
        
        <li class="menu-item menu-item-categories">
          <a href="/categories" rel="section">
            
              <i class="menu-item-icon fa fa-fw fa-th"></i> <br />
            
            分类
          </a>
        </li>
      
        
        <li class="menu-item menu-item-about">
          <a href="/about" rel="section">
            
              <i class="menu-item-icon fa fa-fw fa-user"></i> <br />
            
            关于
          </a>
        </li>
      
        
        <li class="menu-item menu-item-archives">
          <a href="/archives" rel="section">
            
              <i class="menu-item-icon fa fa-fw fa-archive"></i> <br />
            
            归档
          </a>
        </li>
      
        
        <li class="menu-item menu-item-tags">
          <a href="/tags" rel="section">
            
              <i class="menu-item-icon fa fa-fw fa-tags"></i> <br />
            
            标签
          </a>
        </li>
      

      
    </ul>
  

  
</nav>

 </div>
    </header>

    <main id="main" class="main">
      <div class="main-inner">
        <div class="content-wrap">
          <div id="content" class="content">
            

  <div id="posts" class="posts-expand">
    

  
  

  
  
  

  <article class="post post-type-normal " itemscope itemtype="http://schema.org/Article">

    
      <header class="post-header">

        
        
          <h1 class="post-title" itemprop="name headline">
            
            
              
                从源码分析ZeroMQ
              
            
          </h1>
        

        <div class="post-meta">
          <span class="post-time">
            <span class="post-meta-item-icon">
              <i class="fa fa-calendar-o"></i>
            </span>
            <span class="post-meta-item-text">发表于</span>
            <time itemprop="dateCreated" datetime="2016-08-28T15:36:12+08:00" content="2016-08-28">
              2016-08-28
            </time>
          </span>

          
            <span class="post-category" >
              &nbsp; | &nbsp;
              <span class="post-meta-item-icon">
                <i class="fa fa-folder-o"></i>
              </span>
              <span class="post-meta-item-text">分类于</span>
              
                <span itemprop="about" itemscope itemtype="https://schema.org/Thing">
                  <a href="/categories/mq/" itemprop="url" rel="index">
                    <span itemprop="name">mq</span>
                  </a>
                </span>

                
                

              
            </span>
          

          
            
              <span class="post-comments-count">
                &nbsp; | &nbsp;
                <a href="/2016/08/28/ZeroMQ/#comments" itemprop="discussionUrl">
                  <span class="post-comments-count ds-thread-count" data-thread-key="2016/08/28/ZeroMQ/" itemprop="commentsCount"></span>
                </a>
              </span>
            
          

          

          
          

          
        </div>
      </header>
    


    <div class="post-body" itemprop="articleBody">

      
      

      
        <h1 id="ZeroMQ的研究与学习"><a href="#ZeroMQ的研究与学习" class="headerlink" title="ZeroMQ的研究与学习"></a>ZeroMQ的研究与学习</h1><hr>
<blockquote>
<ul>
<li>简介</li>
<li>工作模式</li>
<li>层级模型</li>
<li>实现原理</li>
<li>核心特点</li>
<li>与其他MQ的简单比较</li>
</ul>
</blockquote>
<p><img src="http://img0.imgtn.bdimg.com/it/u=2368283034,2384299468&amp;fm=21&amp;gp=0.jpg" alt="ZeroMQ-logo"></p>
<hr>
<a id="more"></a>
<h2 id="ZeroMQ的一百字概括"><a href="#ZeroMQ的一百字概括" class="headerlink" title="ZeroMQ的一百字概括"></a>ZeroMQ的一百字概括</h2><p><strong>ZeroMQ</strong>看起来想一个可嵌入的网络库，但其作用就像是一个并发框架。它为你提供了各种传输工具，如进程内，进程间，TCP和组播中进行原子消息传递的套接字。你可以使用各种模式实现N对N的套接字连接，这些模式包括发布订阅，请求应答，扇出模式，管道模式。它的速度足够快，因此可以充当集群产品的结构，他的异步IO模型提供了可扩展的多核应用程序，用异步消息来处理任务。它虽然是以C为源码进行开发，但是可以绑定多种语言。</p>
<h3 id="1-简介"><a href="#1-简介" class="headerlink" title="1. 简介"></a>1. 简介</h3><pre><code>ZeroMQ号称是“史上最快的消息队列”，基于c语言开发的，实时流处理sorm的task之间的通信就是用的zeroMQ。
引用官方说法，“ZMQ(以下ZeroMQ简称ZMQ)是一个简单好用的传输层，像框架一样的一个socket library，他使得Socket编程更加简单、简洁和性能更高。是一个消息处理队列库，可在多个线程、内核和主机盒之间弹性伸缩。ZMQ的明确目标是“成为标准网络协议栈的一部分，之后进入Linux内核”。现在还未看到它们的成功。但是，它无疑是极具前景的、并且是人们更加需要的“传统”BSD套接字之上的一 层封装。ZMQ让编写高性能网络应用程序极为简单和有趣。” 确实，它跟RabbitMQ，ActiveMQ之类有着相当本质的区别，ZeroMQ根本就不是一个消息队列服务器，更像是一组底层网络通讯库，对原有的Socket API加上一层封装，是我们操作更简便。使用时只需要引入相应的jar包即可。
</code></pre><h3 id="2-工作模式"><a href="#2-工作模式" class="headerlink" title="2. 工作模式"></a>2. 工作模式</h3><pre><code>ZeroMQ与其他MQ类似，也实现了3中最基本的工作模式：发布-订阅，请求-应答，管道
</code></pre><p>1.发布-订阅<br><img src="http://pic004.cnblogs.com/news/201208/20120813_221017_1.jpg" alt="pub-sub"></p>
<p>“发布-订阅”模式下，“发布者”绑定一个指定的地址，例如“192.168.10.1：5500”，“订阅者”连接到该地址。该模式下消息流是单向的，只允许从“发布者”流向“订阅者”。且“发布者”只管发消息，不理会是否存在“订阅者”。上图只是“发布-订阅”的最基本的模型，一个“发布者”可以拥有多个订阅者，同样的，一个“订阅者”也可订阅多个发布者。下面给出“发布-订阅”模型的样例程序：</p>
<p>发布者：<br><figure class="highlight java"><table><tr><td class="gutter"><pre><div class="line">1</div><div class="line">2</div><div class="line">3</div><div class="line">4</div><div class="line">5</div><div class="line">6</div><div class="line">7</div><div class="line">8</div><div class="line">9</div><div class="line">10</div><div class="line">11</div><div class="line">12</div><div class="line">13</div><div class="line">14</div><div class="line">15</div><div class="line">16</div><div class="line">17</div><div class="line">18</div><div class="line">19</div></pre></td><td class="code"><pre><div class="line"> <span class="keyword">import</span> org.zeromq.ZMQ;  </div><div class="line">  </div><div class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">Publisher</span> </span>&#123;  </div><div class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">void</span> <span class="title">main</span><span class="params">(String args[])</span> </span>&#123;  </div><div class="line">      </div><div class="line">        ZMQ.Context context = ZMQ.context(<span class="number">1</span>);  <span class="comment">//创创建包含一个I/O线程的context  </span></div><div class="line">        ZMQ.Socket publisher = context.socket(ZMQ.PUB);   <span class="comment">//创建一个publisher类型的socket，他可以向所有订阅的subscriber广播数据  </span></div><div class="line">          </div><div class="line">        publisher.bind(<span class="string">"tcp://*:5555"</span>);  <span class="comment">//将当前publisher绑定到5555端口上，可以接受subscriber的订阅  </span></div><div class="line">          </div><div class="line">        <span class="keyword">while</span> (!Thread.currentThread ().isInterrupted ()) &#123;  </div><div class="line">            String message = <span class="string">"fjs hello"</span>;  <span class="comment">//最开始可以理解为pub的channel，subscribe需要订阅fjs这个channel才能接收到消息  </span></div><div class="line">            publisher.send(message.getBytes());  </div><div class="line">        &#125;  </div><div class="line">  </div><div class="line">        publisher.close();  </div><div class="line">        context.term();  </div><div class="line">    &#125;  </div><div class="line">&#125;</div></pre></td></tr></table></figure></p>
<p>订阅者<br><figure class="highlight java"><table><tr><td class="gutter"><pre><div class="line">1</div><div class="line">2</div><div class="line">3</div><div class="line">4</div><div class="line">5</div><div class="line">6</div><div class="line">7</div><div class="line">8</div><div class="line">9</div><div class="line">10</div><div class="line">11</div><div class="line">12</div><div class="line">13</div><div class="line">14</div><div class="line">15</div><div class="line">16</div><div class="line">17</div><div class="line">18</div><div class="line">19</div><div class="line">20</div><div class="line">21</div><div class="line">22</div><div class="line">23</div><div class="line">24</div><div class="line">25</div><div class="line">26</div><div class="line">27</div><div class="line">28</div></pre></td><td class="code"><pre><div class="line"> <span class="keyword">import</span> org.zeromq.ZMQ;  </div><div class="line">  </div><div class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">Subscriber</span> </span>&#123;  </div><div class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">void</span> <span class="title">main</span><span class="params">(String args[])</span> </span>&#123;  </div><div class="line">        <span class="keyword">for</span> (<span class="keyword">int</span> j = <span class="number">0</span>; j &lt; <span class="number">100</span>; j++) &#123;  </div><div class="line">            <span class="keyword">new</span> Thread(<span class="keyword">new</span> Runnable()&#123;  </div><div class="line">  </div><div class="line">                <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">run</span><span class="params">()</span> </span>&#123;  </div><div class="line">                    <span class="comment">// TODO Auto-generated method stub  </span></div><div class="line">                    ZMQ.Context context = ZMQ.context(<span class="number">1</span>);  <span class="comment">//创建1个I/O线程的上下文  </span></div><div class="line">                    ZMQ.Socket subscriber = context.socket(ZMQ.SUB);     <span class="comment">//创建一个sub类型，也就是subscriber类型的socket  </span></div><div class="line">                    subscriber.connect(<span class="string">"tcp://127.0.0.1:5555"</span>);    <span class="comment">//与在5555端口监听的publisher建立连接  </span></div><div class="line">                    subscriber.subscribe(<span class="string">"fjs"</span>.getBytes());     <span class="comment">//订阅fjs这个channel  </span></div><div class="line">                      </div><div class="line">                    <span class="keyword">for</span> (<span class="keyword">int</span> i = <span class="number">0</span>; i &lt; <span class="number">100</span>; i++) &#123;  </div><div class="line">                        <span class="keyword">byte</span>[] message = subscriber.recv();  <span class="comment">//接收publisher发送过来的消息  </span></div><div class="line">                        System.out.println(<span class="string">"receive : "</span> + <span class="keyword">new</span> String(message));  </div><div class="line">                    &#125;  </div><div class="line">                    subscriber.close();  </div><div class="line">                    context.term();  </div><div class="line">                &#125;  </div><div class="line">                  </div><div class="line">            &#125;).start();  </div><div class="line">        &#125;  </div><div class="line">          </div><div class="line">          </div><div class="line">    &#125;  </div><div class="line">&#125;</div></pre></td></tr></table></figure></p>
<p>虽然我们知道“发布者”在发送消息时是不关心“订阅者”的存在于否，所以先启动“发布者”，再启动“订阅者”是很容易导致部分消息丢失的。那么可能会提出一个说法“我先启动‘订阅者’，再启动‘发布者’，就能解决这个问题了？” 对于ZeroMQ而言，这种做法也并不能保证100%的可靠性。在ZeroMQ领域中有一个叫做“<strong>慢木匠</strong>”的术语，就是说即使我是先启动了“订阅者”，再启动“发布者”，“订阅者”总是会丢失第一批数据。因为在“订阅者”与端点建立TCP连接时，会包含几毫秒的握手时间，虽然时间短，但是是存在的。再加上ZeroMQ后台IO是以一部方式执行的，所以若不在双方之间施加同步策略，消息丢失是不可避免的。<br>关于“发布-订阅”模式在ZeroMQ中的一些其他特点：<br>1.公平排队，一个“订阅者”连接到多个发布者时，会均衡的从每个“发布者”读取消息，不会出现一个“发布者”淹没其他“发布者”的情况。<br>2.ZMQ3.0以上的版本，过滤规则发生在“发布方”. ZMQ3.0以下的版本，过滤规则发生在“订阅方”。其实也就是处理消息的位置。</p>
<p>2.请求-应答<br><img src="http://pic004.cnblogs.com/news/201208/20120813_221027_1.jpg" alt="req-res"><br>说到“请求-应答”模式，不得不说的就是它的消息流动模型以及数据包装模型。<br>消息流动模型指的是该模式下，必须严格遵守“一问一答”的方式。<br>在源代码中Req存在两个重要的标志位：<br><figure class="highlight java"><table><tr><td class="gutter"><pre><div class="line">1</div><div class="line">2</div></pre></td><td class="code"><pre><div class="line"><span class="keyword">private</span> <span class="keyword">boolean</span> receiving_reply;  <span class="comment">//标志位，如果是ture的话，表示request已经发送了，正在等待reponse  </span></div><div class="line"><span class="keyword">private</span> <span class="keyword">boolean</span> message_begins;   <span class="comment">//如果是true的话，那么表示这里还需要发送第一个标志的空msg</span></div></pre></td></tr></table></figure></p>
<p>发出消息后，若没有收到回复，再发出第二条消息时就会抛出异常。同样的，对于Rep也是，在没有接收到消息前，不允许发出消息。基于此构成“一问一答”的响应模式。</p>
<p>对于消息发送时的具体数据格式，引入两个图作为参照：<br>含有识别码：<br><img src="https://github.com/imatix/zguide/raw/master/images/fig28.png" alt="has-address"><br>不含识别码：<br><img src="https://github.com/imatix/zguide/raw/master/images/fig26.png" alt="no-address"></p>
<p>这种数据结构ZeroMQ称之为“封包”。一个封包由0个或多个“帧”组成。对于“请求-应答”模式，一个元封包一般由2-3个帧组成，可以看出，差别就是第一帧的存在与否。<br>对于含有目标地址的封包，第一帧存放消息接收端的身份识别码，该码在ZeroMQ内部维护的一个Map中作为key，Value是对应的地址。第二帧是一个分隔帧，没有任何意义，仅仅起分隔作用。第三帧是发送的数据。对于这类封包，通常第一帧，也就是识别码需要我们手动指定。<br>相比于前者，不含识别码的封包内的帧的含义还是一样，只是它的识别码直接有ZeroMQ默认生成，无需手动指定。<br>发送时，根据识别码在内存Map中对应的地址，将消息投递过去。<br>示例程序：<br>服务端<br><figure class="highlight java"><table><tr><td class="gutter"><pre><div class="line">1</div><div class="line">2</div><div class="line">3</div><div class="line">4</div><div class="line">5</div><div class="line">6</div><div class="line">7</div><div class="line">8</div><div class="line">9</div><div class="line">10</div><div class="line">11</div><div class="line">12</div><div class="line">13</div><div class="line">14</div><div class="line">15</div><div class="line">16</div><div class="line">17</div><div class="line">18</div><div class="line">19</div><div class="line">20</div><div class="line">21</div><div class="line">22</div><div class="line">23</div><div class="line">24</div><div class="line">25</div><div class="line">26</div><div class="line">27</div></pre></td><td class="code"><pre><div class="line"> <span class="keyword">import</span> org.zeromq.ZMQ;  </div><div class="line">  </div><div class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">Response</span> </span>&#123;  </div><div class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">void</span> <span class="title">main</span> <span class="params">(String[] args)</span> </span>&#123;  </div><div class="line">        ZMQ.Context context = ZMQ.context(<span class="number">1</span>);  <span class="comment">//这个表示创建用于一个I/O线程的context  </span></div><div class="line">          </div><div class="line">        ZMQ.Socket socket = context.socket(ZMQ.REP);  <span class="comment">//创建一个response类型的socket，他可以接收request发送过来的请求，其实可以将其简单的理解为服务端  </span></div><div class="line">        socket.bind (<span class="string">"tcp://*:5555"</span>);    <span class="comment">//绑定端口  </span></div><div class="line">        <span class="keyword">int</span> i = <span class="number">0</span>;  </div><div class="line">        <span class="keyword">int</span> number = <span class="number">0</span>;  </div><div class="line">        <span class="keyword">while</span> (!Thread.currentThread().isInterrupted()) &#123;  </div><div class="line">            i++;  </div><div class="line">            <span class="keyword">if</span> (i == <span class="number">10000</span>) &#123;  </div><div class="line">                i = <span class="number">0</span>;  </div><div class="line">                System.out.println(++number);  </div><div class="line">            &#125;  </div><div class="line">            <span class="keyword">byte</span>[] request = socket.recv();  <span class="comment">//获取request发送过来的数据  </span></div><div class="line">            <span class="comment">//System.out.println("receive : " + new String(request));  </span></div><div class="line">            String response = <span class="string">"world"</span>;  </div><div class="line">            socket.send(response.getBytes());  <span class="comment">//向request端发送数据  ，必须要要request端返回数据，没有返回就又recv，将会出错，这里可以理解为强制要求走完整个request/response流程  </span></div><div class="line">        &#125;  </div><div class="line">        socket.close();  <span class="comment">//先关闭socket  </span></div><div class="line">        context.term();  <span class="comment">//关闭当前的上下文  </span></div><div class="line">          </div><div class="line">         </div><div class="line">    &#125;  </div><div class="line">&#125;</div></pre></td></tr></table></figure></p>
<p>客户端<br><figure class="highlight java"><table><tr><td class="gutter"><pre><div class="line">1</div><div class="line">2</div><div class="line">3</div><div class="line">4</div><div class="line">5</div><div class="line">6</div><div class="line">7</div><div class="line">8</div><div class="line">9</div><div class="line">10</div><div class="line">11</div><div class="line">12</div><div class="line">13</div><div class="line">14</div><div class="line">15</div><div class="line">16</div><div class="line">17</div><div class="line">18</div><div class="line">19</div><div class="line">20</div><div class="line">21</div><div class="line">22</div><div class="line">23</div><div class="line">24</div><div class="line">25</div><div class="line">26</div><div class="line">27</div><div class="line">28</div><div class="line">29</div><div class="line">30</div></pre></td><td class="code"><pre><div class="line"> <span class="keyword">import</span> org.zeromq.ZMQ;  </div><div class="line">  </div><div class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">Request</span> </span>&#123;  </div><div class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">void</span> <span class="title">main</span><span class="params">(String args[])</span> </span>&#123;  </div><div class="line">        <span class="keyword">for</span> (<span class="keyword">int</span> j = <span class="number">0</span>;  j &lt; <span class="number">5</span>; j++) &#123;  </div><div class="line">            <span class="keyword">new</span> Thread(<span class="keyword">new</span> Runnable()&#123;  </div><div class="line">  </div><div class="line">                <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">run</span><span class="params">()</span> </span>&#123;  </div><div class="line">                    <span class="comment">// TODO Auto-generated method stub  </span></div><div class="line">                    ZMQ.Context context = ZMQ.context(<span class="number">1</span>);  <span class="comment">//创建一个I/O线程的上下文  </span></div><div class="line">                    ZMQ.Socket socket = context.socket(ZMQ.REQ);   <span class="comment">//创建一个request类型的socket，这里可以将其简单的理解为客户端，用于向response端发送数据  </span></div><div class="line">                      </div><div class="line">                    socket.connect(<span class="string">"tcp://127.0.0.1:5555"</span>);   <span class="comment">//与response端建立连接  </span></div><div class="line">                    <span class="keyword">long</span> now = System.currentTimeMillis();  </div><div class="line">                    <span class="keyword">for</span> (<span class="keyword">int</span> i = <span class="number">0</span>; i &lt; <span class="number">100000</span>; i++) &#123;  </div><div class="line">                        String request = <span class="string">"hello"</span>;  </div><div class="line">                        socket.send(request.getBytes());   <span class="comment">//向reponse端发送数据  </span></div><div class="line">                        <span class="keyword">byte</span>[] response = socket.recv();   <span class="comment">//接收response发送回来的数据  正在request/response模型中，send之后必须要recv之后才能继续send，这可能是为了保证整个request/response的流程走完  </span></div><div class="line">                    <span class="comment">//  System.out.println("receive : " + new String(response));  </span></div><div class="line">                    &#125;  </div><div class="line">                    <span class="keyword">long</span> after = System.currentTimeMillis();  </div><div class="line">                      </div><div class="line">                    System.out.println((after - now) / <span class="number">1000</span>);  </div><div class="line">                &#125;  </div><div class="line">                  </div><div class="line">            &#125;).start();;  </div><div class="line">        &#125;  </div><div class="line">          </div><div class="line">    &#125;  </div><div class="line">&#125;</div></pre></td></tr></table></figure></p>
<p>“请求-应答”模式中Req套接字是同步的，每次只跟一个节点交流，如果Req套接字连接了多个节点，请求会同时分发到每一个节点。<br>相应的Rep套接字也是同步的，每次只跟一个节点交流，如果连接了多个节点，则以公平的方式以此从每个节点读取请求，但是最先响应的是最后读取的请求。<br>在接下来的内部结构分析时，将以“请求-应答”模式为例。</p>
<p>3.管道模式<br><img src="http://pic004.cnblogs.com/news/201208/20120813_221017_2.jpg" alt="channel"></p>
<p>在说明“管道模式”前，需要明确的是在ZeroMQ中并没有绝对的服务端与客户端之分，所有的数据接收与发送都是以连接为单位的，只区分ZeroMQ定义的类型。就像套接字绑定地址时，可以使用“bind”，也可以使用“connect”，只是通常我们将理解中的服务端“bind”到一个地址，而理解中的客户端“connec”到该地址。</p>
<p>“管道模式”一般用于任务分发与结果收集，由一个任务发生器来产生任务，“公平”的派发到其管辖下的所有worker，完成后再由结果收集器来回收任务的执行结果。</p>
<p>任务发生器<br><figure class="highlight java"><table><tr><td class="gutter"><pre><div class="line">1</div><div class="line">2</div><div class="line">3</div><div class="line">4</div><div class="line">5</div><div class="line">6</div><div class="line">7</div><div class="line">8</div><div class="line">9</div><div class="line">10</div><div class="line">11</div><div class="line">12</div><div class="line">13</div><div class="line">14</div><div class="line">15</div><div class="line">16</div><div class="line">17</div></pre></td><td class="code"><pre><div class="line"><span class="keyword">import</span> org.zeromq.ZMQ;  </div><div class="line">  </div><div class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">Push</span> </span>&#123;  </div><div class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">void</span> <span class="title">main</span><span class="params">(String args[])</span> </span>&#123;  </div><div class="line">  </div><div class="line">        ZMQ.Context context = ZMQ.context(<span class="number">1</span>);  </div><div class="line">        ZMQ.Socket push  = context.socket(ZMQ.PUSH);  </div><div class="line">        push.bind(<span class="string">"ipc://fjs"</span>);  </div><div class="line">  </div><div class="line">        <span class="keyword">for</span> (<span class="keyword">int</span> i = <span class="number">0</span>; i &lt; <span class="number">10000000</span>; i++) &#123;  </div><div class="line">            push.send(<span class="string">"hello"</span>.getBytes());  </div><div class="line">        &#125;  </div><div class="line">        push.close();  </div><div class="line">        context.term();  </div><div class="line">      </div><div class="line">    &#125;  </div><div class="line">&#125;</div></pre></td></tr></table></figure></p>
<p>Worker<br><figure class="highlight java"><table><tr><td class="gutter"><pre><div class="line">1</div><div class="line">2</div><div class="line">3</div><div class="line">4</div><div class="line">5</div><div class="line">6</div><div class="line">7</div><div class="line">8</div><div class="line">9</div><div class="line">10</div><div class="line">11</div><div class="line">12</div><div class="line">13</div><div class="line">14</div><div class="line">15</div><div class="line">16</div><div class="line">17</div><div class="line">18</div><div class="line">19</div><div class="line">20</div><div class="line">21</div><div class="line">22</div><div class="line">23</div><div class="line">24</div><div class="line">25</div><div class="line">26</div><div class="line">27</div><div class="line">28</div><div class="line">29</div><div class="line">30</div><div class="line">31</div></pre></td><td class="code"><pre><div class="line"><span class="keyword">import</span> java.util.concurrent.atomic.AtomicInteger;  </div><div class="line">  </div><div class="line"><span class="keyword">import</span> org.zeromq.ZMQ;  </div><div class="line">  </div><div class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">Pull</span> </span>&#123;  </div><div class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">void</span> <span class="title">main</span><span class="params">(String args[])</span> </span>&#123;  </div><div class="line">        <span class="keyword">final</span> AtomicInteger number = <span class="keyword">new</span> AtomicInteger(<span class="number">0</span>);  </div><div class="line">        <span class="keyword">for</span> (<span class="keyword">int</span> i = <span class="number">0</span>; i &lt; <span class="number">5</span>; i++) &#123;  </div><div class="line">            <span class="keyword">new</span> Thread(<span class="keyword">new</span> Runnable()&#123;  </div><div class="line">                <span class="keyword">private</span> <span class="keyword">int</span> here = <span class="number">0</span>;  </div><div class="line">                <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">run</span><span class="params">()</span> </span>&#123;  </div><div class="line">                    <span class="comment">// TODO Auto-generated method stub  </span></div><div class="line">                    ZMQ.Context context = ZMQ.context(<span class="number">1</span>);  </div><div class="line">                    ZMQ.Socket pull = context.socket(ZMQ.PULL);  </div><div class="line">                    pull.connect(<span class="string">"ipc://fjs"</span>);  </div><div class="line">                    <span class="comment">//pull.connect("ipc://fjs");  </span></div><div class="line">                    <span class="keyword">while</span> (<span class="keyword">true</span>) &#123;  </div><div class="line">                        String message = <span class="keyword">new</span> String(pull.recv());  </div><div class="line">                        <span class="keyword">int</span> now = number.incrementAndGet();  </div><div class="line">                        here++;  </div><div class="line">                        <span class="keyword">if</span> (now % <span class="number">1000000</span> == <span class="number">0</span>) &#123;  </div><div class="line">                            System.out.println(now + <span class="string">"  here is : "</span> + here);  </div><div class="line">                        &#125;  </div><div class="line">                    &#125;  </div><div class="line">                &#125;  </div><div class="line">                  </div><div class="line">            &#125;).start();  </div><div class="line">              </div><div class="line">        &#125;  </div><div class="line">    &#125;  </div><div class="line">&#125;</div></pre></td></tr></table></figure></p>
<p>结果收集器<br><figure class="highlight java"><table><tr><td class="gutter"><pre><div class="line">1</div><div class="line">2</div><div class="line">3</div><div class="line">4</div><div class="line">5</div><div class="line">6</div><div class="line">7</div><div class="line">8</div><div class="line">9</div><div class="line">10</div><div class="line">11</div><div class="line">12</div><div class="line">13</div><div class="line">14</div><div class="line">15</div><div class="line">16</div><div class="line">17</div><div class="line">18</div><div class="line">19</div></pre></td><td class="code"><pre><div class="line"><span class="keyword">import</span> org.zeromq.ZMQ;  </div><div class="line">  </div><div class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">Pull</span> </span>&#123;  </div><div class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">void</span> <span class="title">main</span><span class="params">(String args[])</span> </span>&#123;  </div><div class="line">        ZMQ.Context context = ZMQ.context(<span class="number">1</span>);  </div><div class="line">        ZMQ.Socket pull = context.socket(ZMQ.PULL);  </div><div class="line">          </div><div class="line">        pull.bind(<span class="string">"ipc://fjs"</span>);  </div><div class="line">          </div><div class="line">        <span class="keyword">int</span> number = <span class="number">0</span>;  </div><div class="line">        <span class="keyword">while</span> (<span class="keyword">true</span>) &#123;  </div><div class="line">            String message = <span class="keyword">new</span> String(pull.recv());  </div><div class="line">            number++;  </div><div class="line">            <span class="keyword">if</span> (number % <span class="number">1000000</span> == <span class="number">0</span>) &#123;  </div><div class="line">                System.out.println(number);  </div><div class="line">            &#125;  </div><div class="line">        &#125;  </div><div class="line">    &#125;  </div><div class="line">&#125;</div></pre></td></tr></table></figure></p>
<p>整体流程比较好理解，Worker连接到任务发生器上，等待任务的产生，完成后将结果发送至结果收集器。如果要以客户端服务端的概念来区分，这里的任务发生器与结果收集器是服务端，而worker是客户端。<br>前面说到了这里任务的派发是“公平的”，因为内部采用了LRU的算法来找到最近最久未工作的闲置worker。但是公平在这里是相对的，当任务发生器启动后，第一个连接到它的worker会在一瞬间承受整个任务发生器产生的tasks。</p>
<h3 id="3-层级模型与交互逻辑"><a href="#3-层级模型与交互逻辑" class="headerlink" title="3. 层级模型与交互逻辑"></a>3. 层级模型与交互逻辑</h3><p><img src="http://img.blog.csdn.net/20160530140759747" alt="层级模型"></p>
<p>这是ZeroMQ的主要的层级模型，以“请求-应答”为例。<br>由上而下，最顶层的是ZObject与IPollEvent。<br><strong>ZObject</strong>是所有ZeroMQ体系中类的父类，它存在的意义是发送与接收命令（有别于消息，命令是告诉ZeroMQ该做什么，需要做什么）。<br><strong>IPollEvent</strong>则是一个接口，定义了若干操作，包括读操作，写操作，客户端请求连接，服务端应答连接，超时操作等供5个操作，该操作的实现类包括Req，Rep等具体Socket。该接口的目的是定义终端间发生操作时的行为。<br><strong>Ctx</strong>是一个上下文类，通常一个终端只需要创建一个上下文。<br><strong>IOObject</strong>本身并没有太多的属性，主要是其内维护了一个<strong>IOThread</strong>。<br><strong>IOThread</strong>是用于处理命令的一个类，内部持有一个MailBox实例与Poller实例。<br><strong>MailBox</strong>是一个重要的类，它被用作处理命令，包括命令的发送与接收，需要注意的是，这里的命令其实是本地发送，也就是自己跟自己发，而不是端点间发送。<br><strong>Pipe</strong>用于处理接收到或者需要发送的数据，是实际存储待处理数据的数据结构，其内部是用队列的形式实现。<br><strong>LB</strong>、<strong>FQ</strong>这两者在官方给出的全名是“LoadBalance”，“FairQueue”。也就是负载均衡与公平排队，分别用于处理要发送的数据与要接收的数据。<br><strong>SocketBase</strong>是例如Req，Rep，Pull等包装后Socket的父类。其内含有一对Pipe（与SessionBase公用），用于在SocketBase与SessionBase之间传递消息，具体传递过程在接下去或说明。<br><strong>SessionBase</strong>是创建SocketChannel并与目标终端进行连接的地方，是与底层Poller最先进行交互的一层。具有超时重连，断线重连等功能。<br><strong>Poller</strong>是整个ZeroMQ的核心，它实现了命令的发送与接收，数据的发送与接收。由他来真正的发送数据到其他终端，也是他处理来自其他终端的数据后交给SessionBase。</p>
<p>基于此层级模型的交互逻辑：</p>
<p>发送消息<br>Socket -&gt; Session -&gt; StreamEngine -&gt; Poller<br>接收消息<br>Poller -&gt; StreamEngine -&gt; Session -&gt; Socket</p>
<h3 id="4-实现原理"><a href="#4-实现原理" class="headerlink" title="4. 实现原理"></a>4. 实现原理</h3><p>这部分将说明从创建一个Socket开始到发送或者接收数据的整个过程，ZeroMQ内部的处理流程。<br>不过我个人觉得先了解一些在底层的原理，对于整体的实现理解会有更好的帮助。</p>
<p>先看一下Poller的一些重要定义<br><figure class="highlight java"><table><tr><td class="gutter"><pre><div class="line">1</div><div class="line">2</div><div class="line">3</div><div class="line">4</div><div class="line">5</div><div class="line">6</div><div class="line">7</div><div class="line">8</div><div class="line">9</div><div class="line">10</div><div class="line">11</div><div class="line">12</div><div class="line">13</div><div class="line">14</div><div class="line">15</div><div class="line">16</div><div class="line">17</div><div class="line">18</div><div class="line">19</div><div class="line">20</div><div class="line">21</div><div class="line">22</div><div class="line">23</div></pre></td><td class="code"><pre><div class="line"><span class="keyword">private</span> <span class="keyword">static</span> <span class="class"><span class="keyword">class</span> <span class="title">PollSet</span> </span>&#123;  </div><div class="line">      <span class="keyword">protected</span> IPollEvents handler;   <span class="comment">//事件的回调  </span></div><div class="line">      <span class="keyword">protected</span> SelectionKey key;   <span class="comment">//注册之后的key  </span></div><div class="line">      <span class="keyword">protected</span> <span class="keyword">int</span> ops;    <span class="comment">//注册的事件  </span></div><div class="line">      <span class="keyword">protected</span> <span class="keyword">boolean</span> cancelled;   <span class="comment">//是否已经取消  </span></div><div class="line">        </div><div class="line">      <span class="function"><span class="keyword">protected</span> <span class="title">PollSet</span><span class="params">(IPollEvents handler)</span> </span>&#123;  </div><div class="line">          <span class="keyword">this</span>.handler = handler;  </div><div class="line">          key = <span class="keyword">null</span>;  </div><div class="line">          cancelled = <span class="keyword">false</span>;  </div><div class="line">          ops = <span class="number">0</span>;  </div><div class="line">      &#125;  </div><div class="line">  &#125;  </div><div class="line">  <span class="keyword">final</span> <span class="keyword">private</span> Map&lt;SelectableChannel, PollSet&gt; fd_table;   <span class="comment">//记录所有的注册，key是channel  </span></div><div class="line"> </div><div class="line">  <span class="keyword">private</span> <span class="keyword">boolean</span> retired;    <span class="comment">//当前注册的对象是否有更新，如果有更新的话，在执行select之前需要先更新注册  </span></div><div class="line"></div><div class="line">  <span class="keyword">volatile</span> <span class="keyword">private</span> <span class="keyword">boolean</span> stopping;    <span class="comment">//如果是true的话，那么执行线程将会停止  </span></div><div class="line">  <span class="keyword">volatile</span> <span class="keyword">private</span> <span class="keyword">boolean</span> stopped;   <span class="comment">//是否已经停止  </span></div><div class="line">    </div><div class="line">  <span class="keyword">private</span> Thread worker;   <span class="comment">//worker线程  </span></div><div class="line">  <span class="keyword">private</span> Selector selector;   <span class="comment">//selector  </span></div><div class="line">  <span class="keyword">final</span> <span class="keyword">private</span> String name;   <span class="comment">//名字</span></div></pre></td></tr></table></figure></p>
<p>PollerSet是Poller的一个嵌套类，所有需要注册到selector上的channel都会先构建这个对象，将其当做附件注册到selector上。其中handler是事件回调（一般是一个IOObject实例），key是selector注册后取得的key，ops是注册的事件类型</p>
<p>fd_table用于维护注册的channel对象与其的PollSet对象之间的映射关系。</p>
<p>retired用于标识当前的注册的channel什么的是否有更新，若是需要更新，则可能会重新生成key。</p>
<p>接下来来看看如何在poller对象上面注册channel吧，有几个比较重要的方法：</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><div class="line">1</div><div class="line">2</div><div class="line">3</div><div class="line">4</div><div class="line">5</div><div class="line">6</div><div class="line">7</div><div class="line">8</div><div class="line">9</div><div class="line">10</div><div class="line">11</div><div class="line">12</div><div class="line">13</div><div class="line">14</div><div class="line">15</div><div class="line">16</div><div class="line">17</div><div class="line">18</div><div class="line">19</div><div class="line">20</div><div class="line">21</div><div class="line">22</div><div class="line">23</div></pre></td><td class="code"><pre><div class="line"></div><div class="line"><span class="comment">//用于在当前的集合里面添加需要注册的channel，第一个参数是channel，第二个参数是事件回调  </span></div><div class="line"><span class="function"><span class="keyword">public</span> <span class="keyword">final</span> <span class="keyword">void</span> <span class="title">add_fd</span> <span class="params">(SelectableChannel fd_, IPollEvents events_)</span> </span>&#123;  </div><div class="line">    fd_table.put(fd_, <span class="keyword">new</span> PollSet(events_));  <span class="comment">//直接把放到map里面就好了  </span></div><div class="line">    adjust_load (<span class="number">1</span>);  <span class="comment">//增加load值，这里所谓的负载其实就是在当前poller里面注册的channel的数量  </span></div><div class="line">&#125;  </div><div class="line"><span class="comment">//在key上面注册事件，如果negate为true的话，那么表示是取消事件  </span></div><div class="line"><span class="function"><span class="keyword">private</span> <span class="keyword">final</span> <span class="keyword">void</span> <span class="title">register</span> <span class="params">(SelectableChannel handle_, <span class="keyword">int</span> ops, <span class="keyword">boolean</span> negate)</span> </span>&#123;  </div><div class="line">    PollSet pollset = fd_table.get(handle_);  <span class="comment">//获取pollset对象  </span></div><div class="line">      </div><div class="line">    <span class="keyword">if</span> (negate)  &#123;  </div><div class="line">        pollset.ops = pollset.ops &amp;~ ops;  <span class="comment">//取反，相当于取消事件  </span></div><div class="line">    &#125; <span class="keyword">else</span> &#123;  </div><div class="line">        pollset.ops = pollset.ops | ops;  <span class="comment">//注册事件  </span></div><div class="line">    &#125;  </div><div class="line">      </div><div class="line">    <span class="keyword">if</span> (pollset.key != <span class="keyword">null</span>) &#123;  <span class="comment">//如果有key了，那么表示已经注册到selector上面了，那么只需要更新key就好了  </span></div><div class="line">        pollset.key.interestOps(pollset.ops);    </div><div class="line">    &#125; <span class="keyword">else</span> &#123;  </div><div class="line">        retired = <span class="keyword">true</span>;  </div><div class="line">  </div><div class="line">    &#125;  </div><div class="line">&#125;</div></pre></td></tr></table></figure>
<p>可见在Poller里注册一个事件主要分为两步 1.放入map中 2.设置PollerSet的相应属性</p>
<p>Poller本身作为一个线程，来看看它的run方法</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><div class="line">1</div><div class="line">2</div><div class="line">3</div><div class="line">4</div><div class="line">5</div><div class="line">6</div><div class="line">7</div><div class="line">8</div><div class="line">9</div><div class="line">10</div><div class="line">11</div><div class="line">12</div><div class="line">13</div><div class="line">14</div><div class="line">15</div><div class="line">16</div><div class="line">17</div><div class="line">18</div><div class="line">19</div><div class="line">20</div><div class="line">21</div><div class="line">22</div><div class="line">23</div><div class="line">24</div><div class="line">25</div><div class="line">26</div><div class="line">27</div><div class="line">28</div><div class="line">29</div><div class="line">30</div><div class="line">31</div><div class="line">32</div><div class="line">33</div><div class="line">34</div><div class="line">35</div><div class="line">36</div><div class="line">37</div><div class="line">38</div><div class="line">39</div><div class="line">40</div><div class="line">41</div><div class="line">42</div><div class="line">43</div><div class="line">44</div><div class="line">45</div><div class="line">46</div><div class="line">47</div><div class="line">48</div><div class="line">49</div><div class="line">50</div><div class="line">51</div><div class="line">52</div><div class="line">53</div><div class="line">54</div><div class="line">55</div><div class="line">56</div><div class="line">57</div><div class="line">58</div><div class="line">59</div><div class="line">60</div><div class="line">61</div><div class="line">62</div><div class="line">63</div><div class="line">64</div><div class="line">65</div><div class="line">66</div><div class="line">67</div><div class="line">68</div><div class="line">69</div><div class="line">70</div><div class="line">71</div><div class="line">72</div><div class="line">73</div><div class="line">74</div><div class="line">75</div><div class="line">76</div><div class="line">77</div><div class="line">78</div><div class="line">79</div><div class="line">80</div><div class="line">81</div><div class="line">82</div><div class="line">83</div><div class="line">84</div><div class="line">85</div></pre></td><td class="code"><pre><div class="line"> <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">run</span> <span class="params">()</span> </span>&#123;  </div><div class="line">    <span class="keyword">int</span> returnsImmediately = <span class="number">0</span>;  </div><div class="line">  </div><div class="line">    <span class="keyword">while</span> (!stopping) &#123;  </div><div class="line">        <span class="keyword">long</span> timeout = execute_timers ();  <span class="comment">//执行所有的超时，并且获取下一个超时的时间  </span></div><div class="line">          </div><div class="line">        <span class="keyword">if</span> (retired) &#123;  <span class="comment">//这里表示注册的东西有更新  </span></div><div class="line">              </div><div class="line">            Iterator &lt;Map.Entry &lt;SelectableChannel,PollSet&gt;&gt; it = fd_table.entrySet ().iterator ();  </div><div class="line">            <span class="keyword">while</span> (it.hasNext ()) &#123;  <span class="comment">//遍历所有需要注册的  </span></div><div class="line">                Map.Entry &lt;SelectableChannel,PollSet&gt; entry = it.next ();  </div><div class="line">                SelectableChannel ch = entry.getKey ();  <span class="comment">//获取channel  </span></div><div class="line">                PollSet pollset = entry.getValue ();   <span class="comment">//获取pollset  </span></div><div class="line">                <span class="keyword">if</span> (pollset.key == <span class="keyword">null</span>) &#123;  <span class="comment">//这里没有key的话，表示当前channel并没有注册到selector上面去  </span></div><div class="line">                    <span class="keyword">try</span> &#123;  </div><div class="line">                        pollset.key = ch.register(selector, pollset.ops, pollset.handler);   <span class="comment">//注册 </span></div><div class="line">                    &#125; <span class="keyword">catch</span> (ClosedChannelException e) &#123;  </div><div class="line">                    &#125;  </div><div class="line">                &#125;   </div><div class="line">                  </div><div class="line">                  </div><div class="line">                <span class="keyword">if</span> (pollset.cancelled || !ch.isOpen()) &#123;  <span class="comment">//如果是取消注册，那么直接取消掉就可以了  </span></div><div class="line">                    <span class="keyword">if</span>(pollset.key != <span class="keyword">null</span>) &#123;  </div><div class="line">                        pollset.key.cancel();  </div><div class="line">                    &#125;  </div><div class="line">                    it.remove ();  </div><div class="line">                &#125;  </div><div class="line">            &#125;  </div><div class="line">            retired = <span class="keyword">false</span>;  </div><div class="line">              </div><div class="line">        &#125;  </div><div class="line">  </div><div class="line">        <span class="comment">//  Wait for events.  </span></div><div class="line">        <span class="keyword">int</span> rc;  </div><div class="line">        <span class="keyword">long</span> start = System.currentTimeMillis ();  <span class="comment">//select之前的时间  </span></div><div class="line">        <span class="keyword">try</span> &#123;  </div><div class="line">            rc = selector.select (timeout);  </div><div class="line">        &#125; <span class="keyword">catch</span> (IOException e) &#123;  </div><div class="line">            <span class="keyword">throw</span> <span class="keyword">new</span> ZError.IOException (e);  </div><div class="line">        &#125;  </div><div class="line">          </div><div class="line">        <span class="keyword">if</span> (rc == <span class="number">0</span>) &#123;   <span class="comment">//出错啦，好像  </span></div><div class="line">            <span class="comment">//  Guess JDK epoll bug  </span></div><div class="line">            <span class="keyword">if</span> (timeout == <span class="number">0</span> ||  </div><div class="line">                    System.currentTimeMillis () - start &lt; timeout / <span class="number">2</span>)  </div><div class="line">                returnsImmediately ++;  </div><div class="line">            <span class="keyword">else</span>  </div><div class="line">                returnsImmediately = <span class="number">0</span>;  </div><div class="line">  </div><div class="line">            <span class="keyword">if</span> (returnsImmediately &gt; <span class="number">10</span>) &#123;  </div><div class="line">                rebuildSelector ();   <span class="comment">//重建selector  </span></div><div class="line">                returnsImmediately = <span class="number">0</span>;  </div><div class="line">            &#125;  </div><div class="line">            <span class="keyword">continue</span>;  </div><div class="line">        &#125;  </div><div class="line">  </div><div class="line">  </div><div class="line">        Iterator&lt;SelectionKey&gt; it = selector.selectedKeys().iterator();  <span class="comment">//所有select出来的key  </span></div><div class="line">        <span class="keyword">while</span> (it.hasNext()) &#123;  <span class="comment">//遍历  </span></div><div class="line">            SelectionKey key = it.next();  </div><div class="line">            IPollEvents evt = (IPollEvents) key.attachment();  </div><div class="line">            it.remove();  </div><div class="line">  </div><div class="line">            <span class="keyword">try</span> &#123;  <span class="comment">//接下来就是判断事件的类型执行相应的方法就好了  </span></div><div class="line">                <span class="keyword">if</span> (key.isReadable() ) &#123;  <span class="comment">//有数据可以读取了   </span></div><div class="line">                    evt.in_event();  </div><div class="line">                &#125; <span class="keyword">else</span> <span class="keyword">if</span> (key.isAcceptable()) &#123;  <span class="comment">//有新的连接进来了  </span></div><div class="line">                    evt.accept_event();  </div><div class="line">                &#125; <span class="keyword">else</span> <span class="keyword">if</span> (key.isConnectable()) &#123;  <span class="comment">//连接建立  </span></div><div class="line">                    evt.connect_event();  </div><div class="line">                &#125;   </div><div class="line">                <span class="keyword">if</span> (key.isWritable()) &#123;  <span class="comment">//可写  </span></div><div class="line">                    evt.out_event();  </div><div class="line">                &#125;   </div><div class="line">            &#125; <span class="keyword">catch</span> (CancelledKeyException e) &#123;  </div><div class="line">                <span class="comment">// channel might have been closed  </span></div><div class="line">            &#125;  </div><div class="line">              </div><div class="line">        &#125;  </div><div class="line">  </div><div class="line">    &#125;  </div><div class="line">      </div><div class="line">    stopped = <span class="keyword">true</span>;  </div><div class="line">      </div><div class="line">&#125;</div></pre></td></tr></table></figure>
<p>这部分还是好理解的，首先是检查fd_table是否需要更新，其实就是有没有新插入的channel或者有channel已经失效，由retired标志位决定。如果需要更新，遍历map中每个元素，检查PollerSet里的key，如果没有，则在Selector上进行注册。<br>然后调用selector.select()，若是有事件到来，根据其事件类型以及注册事件时一并传入的handle来决定执行何种操作。<br>简单来说Poller就是一个轮询器，我们在它的Selector上注册相应的channel与事件。而Poller定期扫描来捕获channel的状态。同时我们也了解到一点，Poller才是真正的IO线程持有者。</p>
<p>粗浅的说明了Poller之后，再来看看MailBox</p>
<p>同样，先是介绍一些重要的属性<br><figure class="highlight java"><table><tr><td class="gutter"><pre><div class="line">1</div><div class="line">2</div><div class="line">3</div><div class="line">4</div><div class="line">5</div><div class="line">6</div></pre></td><td class="code"><pre><div class="line"></div><div class="line"><span class="keyword">private</span> <span class="keyword">final</span> YPipe&lt;Command&gt; cpipe;   <span class="comment">//一个用来保存command的队列，内部以链表的形式实现 </span></div><div class="line">  </div><div class="line"><span class="keyword">private</span> <span class="keyword">final</span> Signaler signaler;   <span class="comment">//其实也是一个实现了一个SocketChannel，但是不对外发送消息，而是向Poller发送空白消息，以提醒command队列中有命令需要处理  </span></div><div class="line">  </div><div class="line"><span class="keyword">private</span> <span class="keyword">final</span> Lock sync;  <span class="comment">//只有一个线程从mailbox里面收命令，但是会有很多线程向mialbox里面发送命令，用这个锁来进行同步</span></div></pre></td></tr></table></figure></p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><div class="line">1</div><div class="line">2</div><div class="line">3</div><div class="line">4</div><div class="line">5</div><div class="line">6</div><div class="line">7</div><div class="line">8</div><div class="line">9</div><div class="line">10</div><div class="line">11</div><div class="line">12</div><div class="line">13</div><div class="line">14</div><div class="line">15</div><div class="line">16</div><div class="line">17</div><div class="line">18</div><div class="line">19</div><div class="line">20</div><div class="line">21</div><div class="line">22</div><div class="line">23</div><div class="line">24</div><div class="line">25</div><div class="line">26</div><div class="line">27</div><div class="line">28</div><div class="line">29</div><div class="line">30</div><div class="line">31</div><div class="line">32</div><div class="line">33</div><div class="line">34</div><div class="line">35</div><div class="line">36</div><div class="line">37</div><div class="line">38</div><div class="line">39</div><div class="line">40</div><div class="line">41</div><div class="line">42</div><div class="line">43</div><div class="line">44</div><div class="line">45</div><div class="line">46</div><div class="line">47</div><div class="line">48</div><div class="line">49</div><div class="line">50</div><div class="line">51</div></pre></td><td class="code"><pre><div class="line"></div><div class="line"><span class="function"><span class="keyword">public</span> SelectableChannel <span class="title">get_fd</span> <span class="params">()</span> </span>&#123;  </div><div class="line">    <span class="keyword">return</span> signaler.get_fd ();   <span class="comment">//这里其实获取的是signal用到的pipe的读channel  </span></div><div class="line">&#125;  </div><div class="line">  </div><div class="line"><span class="comment">//向当前的mailbox发送命令，其实就是写到command队列里面去而已  </span></div><div class="line"><span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">send</span> <span class="params">(<span class="keyword">final</span> Command cmd_)</span> </span>&#123;     </div><div class="line">    <span class="keyword">boolean</span> ok = <span class="keyword">false</span>;  </div><div class="line">    sync.lock ();  </div><div class="line">    <span class="keyword">try</span> &#123;  </div><div class="line">        cpipe.write (cmd_, <span class="keyword">false</span>);  </div><div class="line">        ok = cpipe.flush ();  <span class="comment">//pipeflush，这里将会被selector感应到，从而可以执行相应的处理，在执行线程里面执行命令  </span></div><div class="line">    &#125; <span class="keyword">finally</span> &#123;  </div><div class="line">        sync.unlock ();  </div><div class="line">    &#125;  </div><div class="line">      </div><div class="line">    <span class="keyword">if</span> (!ok) &#123;  </div><div class="line">        signaler.send (); <span class="comment">//通过写端写数据，这样子的话会被读端收到  </span></div><div class="line">    &#125;  </div><div class="line">&#125;  </div><div class="line">  </div><div class="line"><span class="comment">//收取命令，如果这里无法立刻获取命令的话，还可以有一个超时时间  </span></div><div class="line"><span class="function"><span class="keyword">public</span> Command <span class="title">recv</span> <span class="params">(<span class="keyword">long</span> timeout_)</span>  </span>&#123;  </div><div class="line">    Command cmd_ = <span class="keyword">null</span>;  </div><div class="line">    <span class="comment">//  Try to get the command straight away.  </span></div><div class="line">    <span class="keyword">if</span> (active) &#123;  </div><div class="line">        cmd_ = cpipe.read ();  <span class="comment">//从队列里面获取命令  </span></div><div class="line">        <span class="keyword">if</span> (cmd_ != <span class="keyword">null</span>) &#123;  </div><div class="line">              </div><div class="line">            <span class="keyword">return</span> cmd_;  </div><div class="line">        &#125;  </div><div class="line">        <span class="comment">//  If there are no more commands available, switch into passive state.  </span></div><div class="line">        active = <span class="keyword">false</span>;  </div><div class="line">        signaler.recv ();  <span class="comment">//这里会从读端不断的读数据  </span></div><div class="line">    &#125;  </div><div class="line">  </div><div class="line">  </div><div class="line">    <span class="comment">//  Wait for signal from the command sender.  </span></div><div class="line">    <span class="keyword">boolean</span> rc = signaler.wait_event (timeout_);  </div><div class="line">    <span class="keyword">if</span> (!rc)  </div><div class="line">        <span class="keyword">return</span> <span class="keyword">null</span>;  </div><div class="line">  </div><div class="line">    <span class="comment">//  We've got the signal. Now we can switch into active state.  </span></div><div class="line">    active = <span class="keyword">true</span>;  </div><div class="line">  </div><div class="line">    <span class="comment">//  Get a command.  </span></div><div class="line">    cmd_ = cpipe.read ();  </div><div class="line">    <span class="keyword">assert</span> (cmd_ != <span class="keyword">null</span>);  </div><div class="line">      </div><div class="line">    <span class="keyword">return</span> cmd_;  </div><div class="line">&#125;</div></pre></td></tr></table></figure>
<p>MailBox，就像之前说过的，只是用于处理命令的一个东西。命令的读写都是对本地的一个队列进行操作。需要注意的是在写命令与读命令之间需要有Signaler来充当信号通知者。<br>Signaler内部有一组变量：<br><figure class="highlight java"><table><tr><td class="gutter"><pre><div class="line">1</div><div class="line">2</div></pre></td><td class="code"><pre><div class="line"><span class="keyword">private</span> Pipe.SinkChannel w;		数据写入端</div><div class="line"><span class="keyword">private</span> Pipe.SourceChannel r;	数据读取端</div></pre></td></tr></table></figure></p>
<p>将SourceChannel 注册到了poller内。 这样，当命令写入到队列中，会触发SinkChannel的write操作，通过SinkChannel向SourceChannel写数据，此时会被poller内的selector感知到。<br>由于IOThred在向poller注册时，传入的回调是“this”，也就是本身，在发生in_event(读事件)时，实际调用的时IOThread的in_event。<br>然后IOThread中的in_event从MailBox中读取数据，实质是从YPipe中读取command。<br>对于Signaler的作用，只是用于提醒poller有命令，它向SinkChannel内写入的数据其实是一个大小为1没有意义的ByteBuffer。只是用于触发在poller内注册的SourceChannel的Readable事件。<br>需要明确的是，command都是针对于本地的。不会在两台不同的机器间传送command，因为send_command并没有走socketchannel，所以不可能通过网络发送。</p>
<p><img src="http://img.blog.csdn.net/20160530185222726" alt="command"><br>MailBox里面的逻辑大致就是如此：<br>1.命令写入YPipe<br>2.Signaler提醒Poller，激活in_event<br>3.MailBox从YPipe读取命令并执行</p>
<p>ok，一些基本的概念说的差不多了，接下来开始说明Socket的创建以及消息的发送过程。</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><div class="line">1</div><div class="line">2</div><div class="line">3</div><div class="line">4</div><div class="line">5</div><div class="line">6</div><div class="line">7</div></pre></td><td class="code"><pre><div class="line"><span class="comment">//这是一个创建上下文，Socket，与目标端进行连接，发送数据以及接收数据的客户端代码</span></div><div class="line">Context context = ZMQ.context(<span class="number">1</span>);</div><div class="line">Socket worker = context.socket(ZMQ.REQ);</div><div class="line">worker.connect(<span class="string">"tcp://localhost:5671"</span>);</div><div class="line">worker.send (<span class="string">"Hi Boss"</span>);</div><div class="line">String workload = worker.recvStr ();</div><div class="line">Sysout.out.println(workload);</div></pre></td></tr></table></figure>
<p>#####1.创建上下文</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><div class="line">1</div><div class="line">2</div><div class="line">3</div><div class="line">4</div><div class="line">5</div><div class="line">6</div><div class="line">7</div><div class="line">8</div><div class="line">9</div><div class="line">10</div><div class="line">11</div><div class="line">12</div><div class="line">13</div><div class="line">14</div><div class="line">15</div><div class="line">16</div><div class="line">17</div></pre></td><td class="code"><pre><div class="line"><span class="keyword">private</span> <span class="keyword">final</span> List&lt;SocketBase&gt; sockets;</div><div class="line"><span class="keyword">private</span> <span class="keyword">final</span> Deque&lt;Integer&gt; empty_slots;</div><div class="line"><span class="keyword">private</span> <span class="keyword">volatile</span> <span class="keyword">boolean</span> starting;</div><div class="line"><span class="keyword">private</span> <span class="keyword">boolean</span> terminating;</div><div class="line"><span class="keyword">private</span> <span class="keyword">final</span> Lock slot_sync;</div><div class="line"><span class="keyword">private</span> Reaper reaper;</div><div class="line"><span class="keyword">private</span> <span class="keyword">final</span> List&lt;IOThread&gt; io_threads;</div><div class="line"><span class="keyword">private</span> <span class="keyword">int</span> slot_count;</div><div class="line"><span class="keyword">private</span> Mailbox[] slots;</div><div class="line"><span class="keyword">private</span> <span class="keyword">final</span> Mailbox term_mailbox;</div><div class="line"><span class="keyword">private</span> <span class="keyword">final</span> Lock endpoints_sync;</div><div class="line"><span class="keyword">private</span> <span class="keyword">static</span> AtomicInteger max_socket_id = <span class="keyword">new</span> AtomicInteger(<span class="number">0</span>);</div><div class="line"><span class="keyword">private</span> <span class="keyword">int</span> max_sockets;</div><div class="line"><span class="keyword">private</span> <span class="keyword">int</span> io_thread_count;</div><div class="line"><span class="keyword">private</span> <span class="keyword">final</span> Lock opt_sync;</div><div class="line"><span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">final</span> <span class="keyword">int</span> term_tid = <span class="number">0</span>;</div><div class="line"><span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">final</span> <span class="keyword">int</span> reaper_tid = <span class="number">1</span>;</div></pre></td></tr></table></figure>
<p>上面给出了Ctx内的一些重要的成员变量，初始化过程中调用了init_ctx()，返回一个Ctx对象，此时仅仅只是对部分成员变量做了一个初始赋值，并没有特殊操作。</p>
<p>#####2.创建Socket<br><figure class="highlight java"><table><tr><td class="gutter"><pre><div class="line">1</div><div class="line">2</div><div class="line">3</div><div class="line">4</div><div class="line">5</div><div class="line">6</div><div class="line">7</div><div class="line">8</div><div class="line">9</div><div class="line">10</div><div class="line">11</div><div class="line">12</div><div class="line">13</div><div class="line">14</div><div class="line">15</div><div class="line">16</div><div class="line">17</div><div class="line">18</div><div class="line">19</div><div class="line">20</div><div class="line">21</div><div class="line">22</div><div class="line">23</div><div class="line">24</div><div class="line">25</div><div class="line">26</div><div class="line">27</div><div class="line">28</div><div class="line">29</div><div class="line">30</div><div class="line">31</div><div class="line">32</div><div class="line">33</div><div class="line">34</div><div class="line">35</div><div class="line">36</div><div class="line">37</div><div class="line">38</div><div class="line">39</div><div class="line">40</div><div class="line">41</div><div class="line">42</div></pre></td><td class="code"><pre><div class="line"><span class="comment">//截取部分代码，基本上能表示整个过程</span></div><div class="line"><span class="keyword">if</span> (starting) &#123;</div><div class="line">    </div><div class="line">                starting = <span class="keyword">false</span>;</div><div class="line">                opt_sync.lock ();</div><div class="line">                <span class="keyword">int</span> mazmq = max_sockets;</div><div class="line">                <span class="keyword">int</span> ios = io_thread_count;</div><div class="line">                opt_sync.unlock ();</div><div class="line">                slot_count = mazmq + ios + <span class="number">2</span>;</div><div class="line">                slots = <span class="keyword">new</span> Mailbox[slot_count];</div><div class="line">                slots [term_tid] = term_mailbox;</div><div class="line">                reaper = <span class="keyword">new</span> Reaper (<span class="keyword">this</span>, reaper_tid);</div><div class="line">                slots [reaper_tid] = reaper.get_mailbox ();</div><div class="line">                reaper.start ();</div><div class="line">                <span class="comment">//以上部分创建的Reaper对象与两个MailBox是作用于上下文销毁的时候处理剩余消息以及释放占用资源。</span></div><div class="line">                <span class="comment">//下面是需要关注的部分，ios是在创建Ctx时我们指定需要创建的IO线程数，通常情况1个就足够了。根据我们指定的数量创建相应的IOThread，每个IOThread都有他子身的MailBox。</span></div><div class="line">                <span class="keyword">for</span> (<span class="keyword">int</span> i = <span class="number">2</span>; i != ios + <span class="number">2</span>; i++) &#123;</div><div class="line">                <span class="comment">//创建IOThread对象的时候会创建一个Poller，以及一个MailBox,同时，将MailBox对应的Signaler的SourceChannel注册到Poller中以监听有无command需要执行。</span></div><div class="line">                    IOThread io_thread = <span class="keyword">new</span> IOThread (<span class="keyword">this</span>, i);</div><div class="line">                    io_threads.add(io_thread);</div><div class="line">                    slots [i] = io_thread.get_mailbox ();</div><div class="line">                    <span class="comment">//启动Poller</span></div><div class="line">                    io_thread.start ();</div><div class="line">                &#125;</div><div class="line">                <span class="keyword">for</span> (<span class="keyword">int</span> i = (<span class="keyword">int</span>) slot_count - <span class="number">1</span>;</div><div class="line">                      i &gt;= (<span class="keyword">int</span>) ios + <span class="number">2</span>; i--) &#123;</div><div class="line">                    empty_slots.add (i);</div><div class="line">                    slots [i] = <span class="keyword">null</span>;</div><div class="line">                &#125;</div><div class="line">                </div><div class="line">            &#125;</div><div class="line">            <span class="comment">//以上为if部分，只会在Ctx已经创建好，第一次创建Socket会进入的分支，由于是第一次创建Socket，所以需要对一些Ctx成员进行初始化。而之后只需要创建每个Socket对应的IOThread以及必要属性即可。</span></div><div class="line">            <span class="keyword">int</span> slot = empty_slots.pollLast();</div><div class="line">            <span class="keyword">int</span> sid = max_socket_id.incrementAndGet();</div><div class="line">            <span class="comment">//这一步比较重要，先创建一个SocketBase，它是所有Socket的父类</span></div><div class="line">            s = SocketBase.create (type_, <span class="keyword">this</span>, slot, sid);</div><div class="line">            <span class="keyword">if</span> (s == <span class="keyword">null</span>) &#123;</div><div class="line">                empty_slots.addLast(slot);</div><div class="line">                <span class="keyword">return</span> <span class="keyword">null</span>;</div><div class="line">            &#125;</div><div class="line">            sockets.add (s);</div><div class="line">            slots [slot] = s.get_mailbox ();</div></pre></td></tr></table></figure></p>
<p>来看看SocketBase.create(type_,this,slot,sid)都做了些什么。<br><figure class="highlight java"><table><tr><td class="gutter"><pre><div class="line">1</div><div class="line">2</div><div class="line">3</div><div class="line">4</div><div class="line">5</div><div class="line">6</div><div class="line">7</div><div class="line">8</div><div class="line">9</div><div class="line">10</div><div class="line">11</div><div class="line">12</div><div class="line">13</div><div class="line">14</div><div class="line">15</div><div class="line">16</div><div class="line">17</div><div class="line">18</div><div class="line">19</div><div class="line">20</div><div class="line">21</div><div class="line">22</div><div class="line">23</div><div class="line">24</div><div class="line">25</div><div class="line">26</div><div class="line">27</div><div class="line">28</div><div class="line">29</div><div class="line">30</div><div class="line">31</div><div class="line">32</div><div class="line">33</div><div class="line">34</div><div class="line">35</div><div class="line">36</div></pre></td><td class="code"><pre><div class="line"><span class="comment">//省略了部分操作，实际是先根据我们要创建的Socket类型调用构造函数，然后在构造函数中用super调用父类也就是SocketBase的构造函数...</span></div><div class="line"><span class="comment">//给出部分重要成员</span></div><div class="line"><span class="keyword">private</span> <span class="keyword">int</span> tag;</div><div class="line">    <span class="keyword">private</span> <span class="keyword">boolean</span> ctx_terminated;</div><div class="line">    <span class="keyword">private</span> <span class="keyword">boolean</span> destroyed;</div><div class="line">    <span class="keyword">private</span> <span class="keyword">final</span> Mailbox mailbox;</div><div class="line">    <span class="keyword">private</span> <span class="keyword">final</span> List&lt;Pipe&gt; pipes;</div><div class="line">    <span class="keyword">private</span> Poller poller;</div><div class="line">    <span class="keyword">private</span> SelectableChannel handle</div><div class="line">    <span class="keyword">private</span> SocketBase monitor_socket;</div><div class="line"></div><div class="line"><span class="function"><span class="keyword">protected</span> <span class="title">SocketBase</span> <span class="params">(Ctx parent_, <span class="keyword">int</span> tid_, <span class="keyword">int</span> sid_)</span> </span></div><div class="line">    &#123;</div><div class="line">        <span class="comment">//调了ZObject的构造函数，因为ZObject是所有类的父类</span></div><div class="line">        <span class="keyword">super</span> (parent_, tid_);</div><div class="line">        tag = <span class="number">0xbaddecaf</span>;</div><div class="line">        ctx_terminated = <span class="keyword">false</span>;</div><div class="line">        destroyed = <span class="keyword">false</span>;</div><div class="line">        last_tsc = <span class="number">0</span>;</div><div class="line">        ticks = <span class="number">0</span>;</div><div class="line">        rcvmore = <span class="keyword">false</span>;</div><div class="line">        monitor_socket = <span class="keyword">null</span>;</div><div class="line">        monitor_events = <span class="number">0</span>;</div><div class="line">        </div><div class="line">        options.socket_id = sid_;</div><div class="line">        </div><div class="line">        endpoints = <span class="keyword">new</span> MultiMap&lt;String, Own&gt;();</div><div class="line">        <span class="comment">//这个pipes在后期会起到非常大的作用</span></div><div class="line">        pipes = <span class="keyword">new</span> ArrayList&lt;Pipe&gt;();</div><div class="line">        <span class="comment">//创建了一个MailBox</span></div><div class="line">        mailbox = <span class="keyword">new</span> Mailbox(<span class="string">"socket-"</span> + sid_);</div><div class="line"></div><div class="line">        errno = <span class="keyword">new</span> ValueReference&lt;Integer&gt;(<span class="number">0</span>);</div><div class="line">        ...</div><div class="line">        <span class="keyword">return</span> s;</div><div class="line">    &#125;</div></pre></td></tr></table></figure></p>
<p>那到这里为止，我们已经获得了所需的Socket，但是需要注意的是现在只是获得Socket，但是该Socket还没有跟地址进行绑定或者链接。</p>
<p>现在说connect部分，这部分比较长，所以分开说明。<br><figure class="highlight java"><table><tr><td class="gutter"><pre><div class="line">1</div><div class="line">2</div><div class="line">3</div><div class="line">4</div><div class="line">5</div><div class="line">6</div><div class="line">7</div><div class="line">8</div><div class="line">9</div><div class="line">10</div><div class="line">11</div><div class="line">12</div><div class="line">13</div><div class="line">14</div><div class="line">15</div><div class="line">16</div><div class="line">17</div><div class="line">18</div><div class="line">19</div><div class="line">20</div></pre></td><td class="code"><pre><div class="line"><span class="comment">//这里就是先去看看有没有需要执行的command，有的话先执行。这样做的目的应该是假如我们关闭了上下文，理论上来说是不再处理任何请求。但是关闭上下文也是一个动作，发出一个command，经过之前对MailBox的讲解，我们知道处理一个command其实是先放到一个队列中，等待Poller的信号在从队列中取出command然后执行。这样如果Poller要处理较多事件时，可能会推迟command的执行，个人认为在ZeroMQ中，command的优先级是大于消息的。所以基本在执行大部分动作前会先去看看队列中有没有待执行的command。以避免command等待过久而不执行的尴尬。</span></div><div class="line"> <span class="keyword">boolean</span> brc = process_commands (<span class="number">0</span>, <span class="keyword">false</span>);</div><div class="line">        <span class="keyword">if</span> (!brc)</div><div class="line">            <span class="keyword">return</span> <span class="keyword">false</span>;</div><div class="line">            </div><div class="line">...</div><div class="line"></div><div class="line"><span class="comment">//这里没什么需要特别说明的，我们知道终端url的形式是像tcp://192.168.1.1:8000 这样的形式存在的。所以这里做的只是获取IP，端口以及协议。</span></div><div class="line">String protocol = uri.getScheme();</div><div class="line">        String address = uri.getAuthority();</div><div class="line">        String path = uri.getPath();</div><div class="line">        <span class="keyword">if</span> (address == <span class="keyword">null</span>)</div><div class="line">            address = path;</div><div class="line"></div><div class="line">        check_protocol (protocol);</div><div class="line">    </div><div class="line">...</div><div class="line"></div><div class="line"><span class="comment">//创建一个与该Socket对应的Session，一个Socket可以绑定多个Session</span></div><div class="line">SessionBase session = SessionBase.create (io_thread, <span class="keyword">true</span>, <span class="keyword">this</span>, options, paddr);</div></pre></td></tr></table></figure></p>
<p>在SessionBase中才是发生SocketChannel对接的地方，下面来看看它做了些什么。</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><div class="line">1</div><div class="line">2</div><div class="line">3</div><div class="line">4</div><div class="line">5</div><div class="line">6</div><div class="line">7</div><div class="line">8</div><div class="line">9</div><div class="line">10</div><div class="line">11</div><div class="line">12</div><div class="line">13</div><div class="line">14</div><div class="line">15</div><div class="line">16</div><div class="line">17</div><div class="line">18</div><div class="line">19</div><div class="line">20</div><div class="line">21</div><div class="line">22</div><div class="line">23</div><div class="line">24</div><div class="line">25</div><div class="line">26</div><div class="line">27</div><div class="line">28</div><div class="line">29</div><div class="line">30</div><div class="line">31</div><div class="line">32</div></pre></td><td class="code"><pre><div class="line"><span class="comment">//与SocketBase类似，也是进行了一些基本的初始化工作</span></div><div class="line">    <span class="keyword">private</span> <span class="keyword">boolean</span> connect;</div><div class="line">    <span class="keyword">private</span> Pipe pipe;</div><div class="line">    <span class="keyword">private</span> <span class="keyword">final</span> Set&lt;Pipe&gt; terminating_pipes;</div><div class="line">    <span class="keyword">private</span> <span class="keyword">boolean</span> incomplete_in;</div><div class="line">    <span class="keyword">private</span> <span class="keyword">boolean</span> pending;</div><div class="line">    <span class="keyword">private</span> SocketBase socket;</div><div class="line">    <span class="keyword">private</span> IOThread io_thread;</div><div class="line">    <span class="keyword">private</span> <span class="keyword">boolean</span> identity_sent;</div><div class="line">    <span class="keyword">private</span> <span class="keyword">boolean</span> identity_received;</div><div class="line">    <span class="keyword">private</span> <span class="keyword">final</span> Address addr;</div><div class="line"></div><div class="line">    <span class="keyword">private</span> IOObject io_object;</div><div class="line"><span class="function"><span class="keyword">public</span> <span class="title">SessionBase</span><span class="params">(IOThread io_thread_, <span class="keyword">boolean</span> connect_,</span></span></div><div class="line">            SocketBase socket_, Options options_, Address addr_) &#123;</div><div class="line">        <span class="keyword">super</span>(io_thread_, options_);</div><div class="line">        io_object = <span class="keyword">new</span> IOObject(io_thread_);</div><div class="line">        </div><div class="line">        connect = connect_;</div><div class="line">        pipe = <span class="keyword">null</span>;</div><div class="line">        incomplete_in = <span class="keyword">false</span>;</div><div class="line">        pending = <span class="keyword">false</span>;</div><div class="line">        engine = <span class="keyword">null</span>;</div><div class="line">        socket = socket_;</div><div class="line">        io_thread = io_thread_;</div><div class="line">        has_linger_timer = <span class="keyword">false</span>;</div><div class="line">        identity_sent = <span class="keyword">false</span>;</div><div class="line">        identity_received = <span class="keyword">false</span>;</div><div class="line">        addr = addr_;</div><div class="line">     </div><div class="line">        terminating_pipes = <span class="keyword">new</span> HashSet &lt;Pipe&gt; ();</div><div class="line">    &#125;</div></pre></td></tr></table></figure>
<p>继续看connect的过程。</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><div class="line">1</div><div class="line">2</div><div class="line">3</div><div class="line">4</div><div class="line">5</div><div class="line">6</div><div class="line">7</div><div class="line">8</div><div class="line">9</div><div class="line">10</div><div class="line">11</div><div class="line">12</div><div class="line">13</div><div class="line">14</div><div class="line">15</div></pre></td><td class="code"><pre><div class="line">...</div><div class="line"><span class="keyword">if</span> (options.delay_attach_on_connect != <span class="number">1</span> || icanhasall) &#123;</div><div class="line">           <span class="comment">//这个parents在回收资源的时候起作用，维护层级关系</span></div><div class="line">            ZObject[] parents = &#123;<span class="keyword">this</span>, session&#125;;</div><div class="line">            Pipe[] pipes = &#123;<span class="keyword">null</span>, <span class="keyword">null</span>&#125;;</div><div class="line">            <span class="comment">//这是一个高水位线数组，由于Socket根据不同类型会存在发送缓冲区，接收缓冲区，或者一个公用缓冲区。虽然ZeroMQ没有持久化操作。但是比如Req套接字，如果在Rep建立连接前就发送消息，实质是不会发出去的，会先缓存在本地发送缓存区。同时，接收缓冲区也一样，如果收到消息还没来的及处理，就会一直对方在接收缓存区中。高水位线的作用就是给缓冲区定义一个大小，防止撑爆内存。</span></div><div class="line">            <span class="keyword">int</span>[] hwms = &#123;options.sndhwm, options.rcvhwm&#125;;</div><div class="line">            <span class="keyword">boolean</span>[] delays = &#123;options.delay_on_disconnect, options.delay_on_close&#125;;</div><div class="line">            <span class="comment">//OK，接下去的3步操作，直接为Socket与Session进行数据交互奠定了基础。我们仔细看下。</span></div><div class="line">            Pipe.pipepair (parents, pipes, hwms, delays);</div><div class="line"></div><div class="line">            attach_pipe (pipes [<span class="number">0</span>], icanhasall);</div><div class="line"></div><div class="line">            session.attach_pipe (pipes [<span class="number">1</span>]);</div><div class="line">        &#125;</div></pre></td></tr></table></figure>
<figure class="highlight java"><table><tr><td class="gutter"><pre><div class="line">1</div><div class="line">2</div><div class="line">3</div><div class="line">4</div><div class="line">5</div><div class="line">6</div><div class="line">7</div><div class="line">8</div><div class="line">9</div><div class="line">10</div><div class="line">11</div><div class="line">12</div><div class="line">13</div><div class="line">14</div><div class="line">15</div><div class="line">16</div></pre></td><td class="code"><pre><div class="line"><span class="function"><span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">void</span> <span class="title">pipepair</span><span class="params">(ZObject[] parents_, Pipe[] pipes_, <span class="keyword">int</span>[] hwms_, <span class="keyword">boolean</span>[] delays_)</span> </span>&#123;</div><div class="line">	   <span class="comment">//从以下代码可以看到一个数据结构，它创建了2个YPipe对象（实质就是链表），然后又创建了2个Pipe，一般的1个Pipe需要一个写端一个读端，在这里这两个Pipe公用了2个YPipe。</span></div><div class="line">	   <span class="comment">//也就是说，现有2个Pipe，分别是A,B；2个YPipe，分别是Y1，Y2 。A,B公用Y1,Y2，Y1作为A的读端，作为B的写端；Y2作为B的读端，作为A的写端。结构图如下</span></div><div class="line">		YPipe&lt;Msg&gt; upipe1 = <span class="keyword">new</span> YPipe&lt;Msg&gt;(Msg.class, Config.message_pipe_granularity.getValue());</div><div class="line">		YPipe&lt;Msg&gt; upipe2 = <span class="keyword">new</span> YPipe&lt;Msg&gt;(Msg.class, Config.message_pipe_granularity.getValue());</div><div class="line">	            </div><div class="line">	    pipes_ [<span class="number">0</span>] = <span class="keyword">new</span> Pipe(parents_ [<span class="number">0</span>], upipe1, upipe2,</div><div class="line">	        hwms_ [<span class="number">1</span>], hwms_ [<span class="number">0</span>], delays_ [<span class="number">0</span>]);</div><div class="line">	    pipes_ [<span class="number">1</span>] = <span class="keyword">new</span> Pipe(parents_ [<span class="number">1</span>], upipe2, upipe1,</div><div class="line">	        hwms_ [<span class="number">0</span>], hwms_ [<span class="number">1</span>], delays_ [<span class="number">1</span>]);</div><div class="line">	     </div><div class="line">	     <span class="comment">//两个Pipe相互持有对方的引用       </span></div><div class="line">	    pipes_ [<span class="number">0</span>].set_peer (pipes_ [<span class="number">1</span>]);</div><div class="line">	    pipes_ [<span class="number">1</span>].set_peer (pipes_ [<span class="number">0</span>]);</div><div class="line"></div><div class="line">	&#125;</div></pre></td></tr></table></figure>
<p><img src="http://img.blog.csdn.net/20131205133912640?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvZmpzbG92ZWpobA==/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/SouthEast" alt="pipe">    </p>
<p>继续看剩下的两个操作<br><figure class="highlight java"><table><tr><td class="gutter"><pre><div class="line">1</div><div class="line">2</div><div class="line">3</div><div class="line">4</div><div class="line">5</div></pre></td><td class="code"><pre><div class="line"><span class="comment">//这两部操作就比较好说，因为我们有2个Pipe，但最后是SocketBase持有一个Pipe，另一个由SessionBase持有。这样，SessionBase才能通过此与SocketBase进行数据交互，而实际上，发送数据或者接收数据就是通过这两个Pipe来流动的。</span></div><div class="line"></div><div class="line"> attach_pipe (pipes [<span class="number">0</span>], icanhasall);</div><div class="line"></div><div class="line"> session.attach_pipe (pipes [<span class="number">1</span>]);</div></pre></td></tr></table></figure></p>
<p>发送时，Pipe1通过LB向YPipe1写入要发送的数据，并发送read_activated命令，传入参数为Pipe2，SessionBase中其实持有的就是Pipe2，所以Pipe2从YPipe读取数据后由StreamEngine发送<br>接收时，StreamEngine将消息写会SessionBase的Pipe，也就是Pipe2，从Pipe2写入数据，其实是写入到YPipe2，然后通知SocketBase中的Pipe1，Pipe1从YPipe2读取数据<br>其实我觉得比如直接把Pipe设计成队列的形式，同样是两个Pipe，SocketBase与SessionBase同时持有双方引用，也能做到一样的功能，也许是因为这样做的话双方要维护的Pipe对的引用数会加倍，所以没有采用这种做法。</p>
<p>继续讲connect的最后一步。<br><figure class="highlight java"><table><tr><td class="gutter"><pre><div class="line">1</div><div class="line">2</div><div class="line">3</div><div class="line">4</div><div class="line">5</div><div class="line">6</div><div class="line">7</div><div class="line">8</div><div class="line">9</div><div class="line">10</div><div class="line">11</div><div class="line">12</div><div class="line">13</div><div class="line">14</div><div class="line">15</div><div class="line">16</div><div class="line">17</div><div class="line">18</div><div class="line">19</div><div class="line">20</div><div class="line">21</div><div class="line">22</div><div class="line">23</div><div class="line">24</div><div class="line">25</div><div class="line">26</div><div class="line">27</div><div class="line">28</div><div class="line">29</div><div class="line">30</div><div class="line">31</div><div class="line">32</div><div class="line">33</div><div class="line">34</div><div class="line">35</div><div class="line">36</div><div class="line">37</div><div class="line">38</div><div class="line">39</div><div class="line">40</div><div class="line">41</div><div class="line">42</div><div class="line">43</div><div class="line">44</div><div class="line">45</div><div class="line">46</div><div class="line">47</div><div class="line">48</div></pre></td><td class="code"><pre><div class="line">add_endpoint (addr_, session);</div><div class="line"></div><div class="line"><span class="comment">//调用了 launch_child (endpoint_);方法</span></div><div class="line"><span class="function"><span class="keyword">protected</span> <span class="keyword">void</span> <span class="title">launch_child</span> <span class="params">(Own object_)</span></span></div><div class="line">   &#123;</div><div class="line">       <span class="comment">//这步是设置层级关系，回收资源时用到</span></div><div class="line">       object_.set_owner (<span class="keyword">this</span>);</div><div class="line">       <span class="comment">//其实质就是发送了一个plug命令，因为传进来的object_是SessionBase，所以plug操作最后由SessionBase来完成。</span></div><div class="line">       send_plug (object_);</div><div class="line"></div><div class="line">       send_own (<span class="keyword">this</span>, object_);</div><div class="line">   &#125;</div><div class="line">   </div><div class="line">   <span class="comment">//SessionBase中的process_plug操作</span></div><div class="line">    <span class="function"><span class="keyword">protected</span> <span class="keyword">void</span> <span class="title">process_plug</span> <span class="params">()</span></span></div><div class="line">   &#123;</div><div class="line">       io_object.set_handler(<span class="keyword">this</span>);</div><div class="line">       <span class="keyword">if</span> (connect)</div><div class="line">           start_connecting (<span class="keyword">false</span>);</div><div class="line">   &#125;</div><div class="line">   <span class="comment">//看到这里似乎要发现了连接操作将要执行的细节了。</span></div><div class="line">   </div><div class="line">    <span class="function"><span class="keyword">private</span> <span class="keyword">void</span> <span class="title">start_connecting</span> <span class="params">(<span class="keyword">boolean</span> wait_)</span></span></div><div class="line">   &#123;</div><div class="line">       <span class="keyword">assert</span> (connect);</div><div class="line">       <span class="keyword">assert</span> (io_thread != <span class="keyword">null</span>);</div><div class="line"></div><div class="line">       <span class="comment">//  Create the connecter object.</span></div><div class="line"></div><div class="line">       <span class="keyword">if</span> (addr.protocol().equals(<span class="string">"tcp"</span>)) &#123;</div><div class="line">           TcpConnecter connecter = <span class="keyword">new</span> TcpConnecter (</div><div class="line">               io_thread, <span class="keyword">this</span>, options, addr, wait_);</div><div class="line">           <span class="comment">//alloc_assert (connecter);</span></div><div class="line">           <span class="comment">//没错，又是 launch_child，不过这次传进去的对象是上面创建的 TcpConnecter。同样的一会来看看 TcpConnecter里的process_plug操作。</span></div><div class="line">           launch_child (connecter);</div><div class="line">           <span class="keyword">return</span>;</div><div class="line">       &#125;</div><div class="line">       </div><div class="line">       <span class="keyword">if</span> (addr.protocol().equals(<span class="string">"ipc"</span>)) &#123;</div><div class="line">           IpcConnecter connecter = <span class="keyword">new</span> IpcConnecter (</div><div class="line">               io_thread, <span class="keyword">this</span>, options, addr, wait_);</div><div class="line">           <span class="comment">//alloc_assert (connecter);</span></div><div class="line">           launch_child (connecter);</div><div class="line">           <span class="keyword">return</span>;</div><div class="line">       &#125;</div><div class="line">       </div><div class="line">       <span class="keyword">assert</span> (<span class="keyword">false</span>);</div><div class="line">   &#125;</div></pre></td></tr></table></figure></p>
<p>简要说明下TcpConnector的重要成员以及部分操作<br><figure class="highlight java"><table><tr><td class="gutter"><pre><div class="line">1</div><div class="line">2</div><div class="line">3</div><div class="line">4</div><div class="line">5</div><div class="line">6</div><div class="line">7</div><div class="line">8</div><div class="line">9</div><div class="line">10</div><div class="line">11</div><div class="line">12</div><div class="line">13</div><div class="line">14</div><div class="line">15</div><div class="line">16</div><div class="line">17</div><div class="line">18</div><div class="line">19</div><div class="line">20</div><div class="line">21</div><div class="line">22</div><div class="line">23</div><div class="line">24</div><div class="line">25</div><div class="line">26</div><div class="line">27</div><div class="line">28</div><div class="line">29</div><div class="line">30</div><div class="line">31</div><div class="line">32</div><div class="line">33</div><div class="line">34</div><div class="line">35</div><div class="line">36</div><div class="line">37</div><div class="line">38</div><div class="line">39</div><div class="line">40</div><div class="line">41</div><div class="line">42</div><div class="line">43</div><div class="line">44</div><div class="line">45</div><div class="line">46</div><div class="line">47</div><div class="line">48</div><div class="line">49</div><div class="line">50</div><div class="line">51</div><div class="line">52</div><div class="line">53</div><div class="line">54</div><div class="line">55</div><div class="line">56</div><div class="line">57</div><div class="line">58</div><div class="line">59</div><div class="line">60</div><div class="line">61</div><div class="line">62</div><div class="line">63</div><div class="line">64</div><div class="line">65</div><div class="line">66</div><div class="line">67</div><div class="line">68</div><div class="line">69</div><div class="line">70</div><div class="line">71</div><div class="line">72</div><div class="line">73</div><div class="line">74</div><div class="line">75</div><div class="line">76</div><div class="line">77</div><div class="line">78</div><div class="line">79</div><div class="line">80</div><div class="line">81</div><div class="line">82</div></pre></td><td class="code"><pre><div class="line">    <span class="keyword">private</span> <span class="keyword">final</span> <span class="keyword">static</span> <span class="keyword">int</span> reconnect_timer_id = <span class="number">1</span>;</div><div class="line">    <span class="keyword">private</span> <span class="keyword">final</span> IOObject io_object;</div><div class="line">    <span class="keyword">private</span> <span class="keyword">final</span> Address addr;</div><div class="line">    <span class="keyword">private</span> SocketChannel handle;</div><div class="line">    <span class="keyword">private</span> <span class="keyword">boolean</span> handle_valid;</div><div class="line">    <span class="keyword">private</span> <span class="keyword">boolean</span> delayed_start;</div><div class="line">    <span class="keyword">private</span> <span class="keyword">boolean</span> timer_started;</div><div class="line">    <span class="keyword">private</span> SessionBase session;</div><div class="line">    <span class="keyword">private</span> <span class="keyword">int</span> current_reconnect_ivl;</div><div class="line">    <span class="keyword">private</span> String endpoint;</div><div class="line">    <span class="keyword">private</span> SocketBase socket;</div><div class="line">    <span class="comment">//观察下发现他持有一个SessionBase引用，以及含有IOThread的一个IOObject</span></div><div class="line"><span class="function"><span class="keyword">public</span> <span class="title">TcpConnecter</span> <span class="params">(IOThread io_thread_,</span></span></div><div class="line">      SessionBase session_, <span class="keyword">final</span> Options options_,</div><div class="line">      <span class="keyword">final</span> Address addr_, <span class="keyword">boolean</span> delayed_start_) &#123;</div><div class="line">        </div><div class="line">        <span class="keyword">super</span> (io_thread_, options_);</div><div class="line">        io_object = <span class="keyword">new</span> IOObject(io_thread_);</div><div class="line">        addr = addr_;</div><div class="line">        handle = <span class="keyword">null</span>; </div><div class="line">        handle_valid = <span class="keyword">false</span>;</div><div class="line">        delayed_start = delayed_start_;</div><div class="line">        timer_started = <span class="keyword">false</span>;</div><div class="line">        session = session_;</div><div class="line">        current_reconnect_ivl = options.reconnect_ivl;</div><div class="line">        </div><div class="line">        <span class="keyword">assert</span> (addr != <span class="keyword">null</span>);</div><div class="line">        endpoint = addr.toString ();</div><div class="line">        socket = session_.get_soket ();</div><div class="line">    &#125;</div><div class="line">    <span class="comment">//接下去，我们要来看TcpConnector里的process_plug</span></div><div class="line">    <span class="function"><span class="keyword">protected</span> <span class="keyword">void</span> <span class="title">process_plug</span> <span class="params">()</span></span></div><div class="line">    &#123;</div><div class="line">        <span class="comment">//设置了回调句柄为TcpConnector</span></div><div class="line">        io_object.set_handler(<span class="keyword">this</span>);</div><div class="line">        <span class="keyword">if</span> (delayed_start)</div><div class="line">            add_reconnect_timer();</div><div class="line">        <span class="keyword">else</span> &#123;</div><div class="line">            start_connecting ();</div><div class="line">        &#125;</div><div class="line">    &#125;</div><div class="line">    </div><div class="line">    <span class="comment">//开始进行Socket连接</span></div><div class="line">     <span class="function"><span class="keyword">private</span> <span class="keyword">void</span> <span class="title">start_connecting</span> <span class="params">()</span></span></div><div class="line">    &#123;</div><div class="line">        <span class="keyword">try</span> &#123;</div><div class="line">        <span class="comment">//open()操作，执行真正的连接</span></div><div class="line">            <span class="keyword">boolean</span> rc = open ();</div><div class="line">            <span class="keyword">if</span> (rc) &#123;</div><div class="line">            <span class="comment">//若是连接成功，那该SocketChannel注册到Poller中</span></div><div class="line">                io_object.add_fd (handle);</div><div class="line">                handle_valid = <span class="keyword">true</span>;</div><div class="line">                <span class="comment">//ok，这里不用说，实际上肯定是执行了事件回调，在之前的process_plug里看到， io_object.set_handler(this);说明在IOObject内的回调句柄就是该TcpConnector，所以这里执行的就是TcpConnector的connect_event()，具体执行在下面有所介绍。</span></div><div class="line">                io_object.connect_event();</div><div class="line">            &#125;</div><div class="line">            <span class="keyword">else</span> &#123;</div><div class="line">            <span class="comment">//进入该分支那明显就是说明连接未成功，采取的策略是延迟一段时间后继续尝试连接。</span></div><div class="line">                io_object.add_fd (handle);</div><div class="line">                handle_valid = <span class="keyword">true</span>;</div><div class="line">                io_object.set_pollconnect (handle);</div><div class="line">                socket.event_connect_delayed (endpoint, -<span class="number">1</span>);</div><div class="line">            &#125;</div><div class="line">        &#125; <span class="keyword">catch</span> (IOException e) &#123;</div><div class="line">            <span class="comment">//  Handle any other error condition by eventual reconnect.</span></div><div class="line">            <span class="keyword">if</span> (handle != <span class="keyword">null</span>)</div><div class="line">                close ();</div><div class="line">            add_reconnect_timer();</div><div class="line">        &#125;</div><div class="line">    &#125;</div><div class="line">    </div><div class="line">    <span class="comment">//到此，终于看到了真正的Socket连接</span></div><div class="line">     <span class="function"><span class="keyword">private</span> <span class="keyword">boolean</span> <span class="title">open</span> <span class="params">()</span> <span class="keyword">throws</span> IOException</span></div><div class="line">    &#123;</div><div class="line">        <span class="keyword">assert</span> (handle == <span class="keyword">null</span>);</div><div class="line">        handle = SocketChannel.open();</div><div class="line">        <span class="comment">//设置为非阻塞模式</span></div><div class="line">        Utils.unblock_socket(handle);</div><div class="line">        <span class="keyword">boolean</span> rc = handle.connect(addr.resolved().address());</div><div class="line"></div><div class="line">        <span class="keyword">return</span> rc;</div><div class="line"></div><div class="line">    &#125;</div></pre></td></tr></table></figure></p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><div class="line">1</div><div class="line">2</div><div class="line">3</div><div class="line">4</div><div class="line">5</div><div class="line">6</div><div class="line">7</div><div class="line">8</div><div class="line">9</div><div class="line">10</div><div class="line">11</div><div class="line">12</div><div class="line">13</div><div class="line">14</div><div class="line">15</div><div class="line">16</div><div class="line">17</div><div class="line">18</div><div class="line">19</div><div class="line">20</div><div class="line">21</div><div class="line">22</div><div class="line">23</div><div class="line">24</div><div class="line">25</div><div class="line">26</div><div class="line">27</div><div class="line">28</div><div class="line">29</div><div class="line">30</div><div class="line">31</div><div class="line">32</div><div class="line">33</div><div class="line">34</div><div class="line">35</div><div class="line">36</div><div class="line">37</div><div class="line">38</div><div class="line">39</div><div class="line">40</div><div class="line">41</div><div class="line">42</div><div class="line">43</div><div class="line">44</div><div class="line">45</div><div class="line">46</div><div class="line">47</div><div class="line">48</div></pre></td><td class="code"><pre><div class="line">这里说明下connect_event()</div><div class="line"><span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">connect_event</span> <span class="params">()</span></span></div><div class="line">    &#123;</div><div class="line">        <span class="keyword">boolean</span> err = <span class="keyword">false</span>;</div><div class="line">        SocketChannel fd = <span class="keyword">null</span>;</div><div class="line">        <span class="keyword">try</span> &#123;</div><div class="line">        <span class="comment">//确认连接是否完成，因为之前将SocketChannel设置成了非阻塞模式，所以我们并不知道连接是否已经成功建立。</span></div><div class="line">            fd = connect ();</div><div class="line">        &#125; <span class="keyword">catch</span> ()&#123;</div><div class="line">        ...</div><div class="line">        &#125;</div><div class="line">        <span class="comment">//这里是将成功连接得到的SocketChannel从Poller中主注销，可能大家觉得奇怪，都成功连接了干啥要注销呢，因为之前注册的时候监听的是connect_event，现在既然成功连接了，当然不再需要监听connect_event了。</span></div><div class="line">        io_object.rm_fd (handle);</div><div class="line">        handle_valid = <span class="keyword">false</span>;</div><div class="line">        </div><div class="line">        <span class="keyword">if</span> (err) &#123;</div><div class="line">          <span class="comment">//这里做的是失败重连机制，具体做法是延迟再尝试，指数式延迟。比如第一次延迟2s，第二次延迟4s...</span></div><div class="line">          <span class="comment">//至于close()执行的目的，那肯定就是关闭由于之前成功连接所打开的一些资源。</span></div><div class="line">            close ();</div><div class="line">            add_reconnect_timer();</div><div class="line">            <span class="keyword">return</span>;</div><div class="line">        &#125;</div><div class="line">        </div><div class="line">        handle = <span class="keyword">null</span>;</div><div class="line">        </div><div class="line">        <span class="keyword">try</span> &#123;</div><div class="line">            </div><div class="line">            Utils.tune_tcp_socket (fd);</div><div class="line">            Utils.tune_tcp_keepalives (fd, options.tcp_keepalive, options.tcp_keepalive_cnt, options.tcp_keepalive_idle, options.tcp_keepalive_intvl);</div><div class="line">        &#125; <span class="keyword">catch</span> (SocketException e) &#123;</div><div class="line">            <span class="keyword">throw</span> <span class="keyword">new</span> RuntimeException(e);</div><div class="line">        &#125;</div><div class="line">        <span class="comment">//StreamEngine这个东西之前没有讲过，不过等下会细讲，它是位于Poller与SessionBase之间的一层，用于数据处理。</span></div><div class="line">        StreamEngine engine = <span class="keyword">null</span>;</div><div class="line">        <span class="keyword">try</span> &#123;</div><div class="line">            engine = <span class="keyword">new</span> StreamEngine (fd, options, endpoint);</div><div class="line">        &#125; <span class="keyword">catch</span> (ZError.InstantiationException e) &#123;</div><div class="line">            socket.event_connect_delayed (endpoint, -<span class="number">1</span>);</div><div class="line">            <span class="keyword">return</span>;</div><div class="line">        &#125;</div><div class="line"></div><div class="line">      <span class="comment">//上面提到了，StreamEngine是用于SessionBase与Poller之间的，那么需要将SessionBase与StreamEngine联系起来，两者才能进行交互</span></div><div class="line">        send_attach (session, engine);</div><div class="line">    <span class="comment">//以下是清扫一些由于connect而留下的未释放的资源。</span></div><div class="line">        terminate ();</div><div class="line"></div><div class="line">        socket.event_connected (endpoint, fd);</div><div class="line">    &#125;</div></pre></td></tr></table></figure>
<p>ok，讲了很多，那到了这里，Socket的连接基本也就完成了。<br>回顾一下这个过程：</p>
<ol>
<li>connect的动作最初是发生在SocketBase的</li>
<li>在SocketBase中它创建了一个SessionBase与之对应</li>
<li>同时创建了一对Pipe，SocketBase与SessionBase各持一个，用来在两者间进行数据交互。</li>
<li>调用了SessionBase中的process_plug方法，创建一个TcpConnector用来将SocketChannel进行连接</li>
<li>又调用了TcpConnector中的process_plug方法开始进行真正的连接</li>
<li>在连接过程中我们看到，它将SocketChannel设置成了非阻塞模式，所以需要在后续检查连接是否完成，当然未完成时它也有重连策略（指数后退再重连）。</li>
<li>连接成功后，注销在Poller中原有的SocketChannel，在重新注册一个，因为原先的SocketChannel监听connect事件。</li>
<li>我们看到，在整个过程中MailBox起到了至关重要的部分，因为无论是plug或者是connect_event()，都是通过命令来执行的，而不是直接通过引用调用。</li>
</ol>
<p>过程大致如上，再来说说这个StreamEngine，之前也没有讲到这个东西。</p>
<blockquote>
<p>关于Poller，SessionBase与StreamEngine的交互（数据接收与发送的处理）<br>通过之前的了解，ZeroMQ自底向上的层级结构是这样的：底层–&gt;Poller–&gt;StreamEngine–&gt;Session–&gt;Socket–&gt;应用程序<br>Poller与StreamEngine的关系相当于是Poller需要发送的数据是由StreamEngine进行编码处理的，Poller接收到的数据是由StreamEngine来解码处理的。<br>首先StreamEngine何时被创建？<br>上面说的Tcp连接过程中，讲到了超时重连机制。如果连接过程中，成功建立连接，则会先删除原先TcpConnector在Poller注册的channel，因为他只针对事件connect，成功建立连接后就没有用了。然后创建StreamEngine对象，没有其余步骤，只是初始化一些基本变量。<br>接下来是重点<br>需要将当前的Session与Engine相互联系，这样Engine接收到数据后处理好之后能放到Pipe中供session获取，发送也是同样道理，所以这里需要发送attach命令.命令发送的过程不再赘述，上面已经讲过了。<br>接收到attach命令后，由SessionBase的process_attach来负责处理，比较重要的是Engine中的plug方法，该方法的作用是将Engine插到Session中，但是比较好理解的说法就是讲Engine与Session相互联系起来，并且在Poller中进行注册。<br>向poller注册的过程就是用将从tcpconnector中成功连接所获取到的SocketChannel注册到Poller中，Engine本身作为回调方法，里面实现了in_event,out_event事件，到这里poller与Engine已经可以交互，poller在轮询发现有输入或者输出事件时，StreamEngine中的in_event,out_event会去处理。</p>
<p>简单来说<br>StreamEngine与SessionBase的交互，是将一个session赋值给StreamEngine中的SessionBase<br>StreamEngine与Poller的交互，是将从TcpConnector中成功连接返回的SocketChannel注册到Poller中，并将本身作为回调事件加入其中。<br>值得一提的是，在StreamEngine中调用plug方法时，不仅仅是连接SessionBase与Poller，也会发生一次“握手”。</p>
</blockquote>
<p>那进行到这里，剩下的就是一个发送与接收的过程了。<br>关于发送过程，入口仍就是SocketBase的send。</p>
<p>代码也有点长，我们还是老办法，分开看。</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><div class="line">1</div><div class="line">2</div><div class="line">3</div><div class="line">4</div><div class="line">5</div><div class="line">6</div><div class="line">7</div><div class="line">8</div><div class="line">9</div><div class="line">10</div><div class="line">11</div><div class="line">12</div><div class="line">13</div><div class="line">14</div><div class="line">15</div><div class="line">16</div><div class="line">17</div></pre></td><td class="code"><pre><div class="line"><span class="comment">//public boolean send (Msg msg_, int flags_)这是方法名，Msg是消息载体，封装了要发送的消息。flags_是一个标志位，表示消息是由多段组成，合并后一起发送或者单独发送</span></div><div class="line"></div><div class="line"><span class="function"><span class="keyword">public</span> <span class="keyword">boolean</span> <span class="title">send</span> <span class="params">(Msg msg_, <span class="keyword">int</span> flags_)</span></span></div><div class="line">    &#123;</div><div class="line">        ...</div><div class="line">        <span class="comment">//之前怎么说的，命令的优先级大于消息，所以在类似发送消息，接收消息之前都会先去看下有没有命令要执行，主要防止一些例如销毁上下文之类的命令被延迟执行。</span></div><div class="line">        <span class="keyword">boolean</span> brc = process_commands (<span class="number">0</span>, <span class="keyword">true</span>);</div><div class="line">        <span class="keyword">if</span> (!brc)</div><div class="line">            <span class="keyword">return</span> <span class="keyword">false</span>;</div><div class="line">            <span class="comment">//没啥意义，设置Msg的初始标志位</span></div><div class="line">        msg_.reset_flags (Msg.more);</div><div class="line">        </div><div class="line">        <span class="keyword">if</span> ((flags_ &amp; ZMQ.ZMQ_SNDMORE) &gt; <span class="number">0</span>)</div><div class="line">            msg_.set_flags (Msg.more);</div><div class="line"></div><div class="line">        <span class="comment">//  尝试发送消息，经过对ZeroMQ的命名方法的研究，发现带有x都是调用其具体实现类方法。</span></div><div class="line">        <span class="keyword">boolean</span> rc = xsend(msg_);</div></pre></td></tr></table></figure>
<p>那就来看下具体是哪个子类来执行这个发送动作的</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><div class="line">1</div><div class="line">2</div><div class="line">3</div><div class="line">4</div><div class="line">5</div><div class="line">6</div><div class="line">7</div><div class="line">8</div><div class="line">9</div><div class="line">10</div><div class="line">11</div><div class="line">12</div><div class="line">13</div><div class="line">14</div><div class="line">15</div><div class="line">16</div><div class="line">17</div><div class="line">18</div><div class="line">19</div><div class="line">20</div><div class="line">21</div><div class="line">22</div><div class="line">23</div><div class="line">24</div><div class="line">25</div><div class="line">26</div><div class="line">27</div><div class="line">28</div><div class="line">29</div><div class="line">30</div></pre></td><td class="code"><pre><div class="line"><span class="function"><span class="keyword">protected</span> <span class="keyword">boolean</span> <span class="title">xsend</span><span class="params">(Msg msg_)</span></span></div><div class="line">   &#123;</div><div class="line">   <span class="comment">//lb就是之前说的LB（LoadBalanced 用来管理发送的消息的）</span></div><div class="line">       <span class="keyword">return</span> lb.send(msg_, errno);</div><div class="line">   &#125;</div><div class="line">   <span class="comment">//Req的父类是Dealer</span></div><div class="line">   </div><div class="line">   <span class="comment">//来看看LB的send</span></div><div class="line">   <span class="function"><span class="keyword">public</span> <span class="keyword">boolean</span> <span class="title">send</span><span class="params">(Msg msg_, ValueReference&lt;Integer&gt; errno)</span> </span>&#123;</div><div class="line">   ...</div><div class="line">    more = msg_.has_more();</div><div class="line">       <span class="keyword">if</span> (!more) &#123;</div><div class="line">       <span class="comment">//flush是对于发送消息的准备</span></div><div class="line">           pipes.get(current).flush ();</div><div class="line">           <span class="keyword">if</span> (active &gt; <span class="number">1</span>)</div><div class="line">               current = (current + <span class="number">1</span>) % active;</div><div class="line">       &#125;</div><div class="line"></div><div class="line">       <span class="keyword">return</span> <span class="keyword">true</span>;</div><div class="line">   &#125;</div><div class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">flush</span> <span class="params">()</span></span></div><div class="line">   &#123;</div><div class="line">       <span class="keyword">if</span> (state == State.terminating)</div><div class="line">           <span class="keyword">return</span>;</div><div class="line"></div><div class="line">       <span class="keyword">if</span> (outpipe != <span class="keyword">null</span> &amp;&amp; !outpipe.flush ()) &#123;</div><div class="line">       <span class="comment">//发送了一个command，告知已经有准备好发送的数据，请读取</span></div><div class="line">           send_activate_read (peer);</div><div class="line">       &#125; </div><div class="line">   &#125;</div></pre></td></tr></table></figure>
<p>之前提过SessionBase与SocketBase通过一对Pipe来进行数据交互，那到这里，发送数据的过程也已经能看出个大概，首先就是将准备好的数据进行封包，Req需要一个身份帧，一个空帧，一个内容帧。然后将封包后的Msg写入SocketBase的Pipe，同时向Poller发送一个命令，告知在SocketBase端的Pipe中有可读消息需要发送。随即执行SessionBase的read_activated方法，目的是从Pipe中读出要发送的数据后发送出去。<br><figure class="highlight java"><table><tr><td class="gutter"><pre><div class="line">1</div><div class="line">2</div><div class="line">3</div><div class="line">4</div><div class="line">5</div><div class="line">6</div><div class="line">7</div><div class="line">8</div><div class="line">9</div><div class="line">10</div><div class="line">11</div><div class="line">12</div><div class="line">13</div></pre></td><td class="code"><pre><div class="line"><span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">read_activated</span><span class="params">(Pipe pipe_)</span> </span></div><div class="line">   &#123;</div><div class="line">       <span class="comment">// Skip activating if we're detaching this pipe</span></div><div class="line">       <span class="keyword">if</span> (pipe != pipe_) &#123;</div><div class="line">           <span class="keyword">assert</span> (terminating_pipes.contains (pipe_));</div><div class="line">           <span class="keyword">return</span>;</div><div class="line">       &#125;</div><div class="line">       </div><div class="line">       <span class="keyword">if</span> (engine != <span class="keyword">null</span>)</div><div class="line">           engine.activate_out ();</div><div class="line">       <span class="keyword">else</span></div><div class="line">           pipe.check_read ();</div><div class="line">   &#125;</div></pre></td></tr></table></figure></p>
<p>之前也讲过真正的数据发送是发生在StreamEngine中的,那么需要来看下activate_out中发生了什么。<br><figure class="highlight java"><table><tr><td class="gutter"><pre><div class="line">1</div><div class="line">2</div><div class="line">3</div><div class="line">4</div><div class="line">5</div><div class="line">6</div><div class="line">7</div><div class="line">8</div><div class="line">9</div><div class="line">10</div><div class="line">11</div><div class="line">12</div><div class="line">13</div><div class="line">14</div><div class="line">15</div><div class="line">16</div><div class="line">17</div><div class="line">18</div><div class="line">19</div><div class="line">20</div><div class="line">21</div><div class="line">22</div><div class="line">23</div><div class="line">24</div><div class="line">25</div><div class="line">26</div><div class="line">27</div></pre></td><td class="code"><pre><div class="line"> <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">out_event</span> <span class="params">()</span> </span></div><div class="line">&#123;</div><div class="line">    ...</div><div class="line">    <span class="comment">//通过SessionBase的Pipe从SocketBase的Pipe中读取数据</span></div><div class="line">      outbuf = encoder.get_data (<span class="keyword">null</span>);</div><div class="line">      outsize = outbuf.remaining();</div><div class="line">      ...</div><div class="line">      <span class="comment">//发送数据</span></div><div class="line">      <span class="keyword">int</span> nbytes = write (outbuf);</div><div class="line">      ...</div><div class="line">&#125;</div><div class="line"><span class="function"><span class="keyword">private</span> <span class="keyword">int</span> <span class="title">write</span> <span class="params">(Transfer buf)</span> </span></div><div class="line">&#123;</div><div class="line">    <span class="keyword">int</span> nbytes = <span class="number">0</span> ;</div><div class="line">    <span class="keyword">try</span> &#123;</div><div class="line">        nbytes = buf.transferTo(handle);</div><div class="line">    &#125; <span class="keyword">catch</span> (IOException e) &#123;</div><div class="line">        <span class="keyword">return</span> -<span class="number">1</span>;</div><div class="line">    &#125;</div><div class="line">    </div><div class="line">    <span class="keyword">return</span> nbytes;</div><div class="line">&#125;</div><div class="line">  <span class="meta">@Override</span></div><div class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">final</span> <span class="keyword">int</span> <span class="title">transferTo</span> <span class="params">(WritableByteChannel s)</span> <span class="keyword">throws</span> IOException </span>&#123;</div><div class="line">        <span class="keyword">return</span> s.write (buf);</div><div class="line">    &#125;</div><div class="line">    <span class="comment">//那看到这里我们已经看到了真正的发送过程，读取数据后，直接通过WritableByteChannel来想连接节点发送消息。</span></div></pre></td></tr></table></figure></p>
<p>接收过程类似于发送过程，Poller轮询器里注册的channel监听到有事件发生，由于之前channel向selector注册的时候携带的附件基本都是一个IOObject对象，所以从IOObject对象开始逐层向内执行in_event()，最终到了StreamEngine中。<br><figure class="highlight java"><table><tr><td class="gutter"><pre><div class="line">1</div><div class="line">2</div><div class="line">3</div><div class="line">4</div><div class="line">5</div><div class="line">6</div><div class="line">7</div><div class="line">8</div><div class="line">9</div><div class="line">10</div><div class="line">11</div></pre></td><td class="code"><pre><div class="line"><span class="function"><span class="keyword">private</span> <span class="keyword">int</span> <span class="title">read</span> <span class="params">(ByteBuffer buf)</span> </span></div><div class="line">    &#123;</div><div class="line">        <span class="keyword">int</span> nbytes = <span class="number">0</span> ;</div><div class="line">        <span class="keyword">try</span> &#123;</div><div class="line">            nbytes = handle.read (buf);</div><div class="line">        &#125; <span class="keyword">catch</span> (IOException e) &#123;</div><div class="line">            <span class="keyword">return</span> -<span class="number">1</span>;</div><div class="line">        &#125;</div><div class="line">        </div><div class="line">        <span class="keyword">return</span> nbytes;</div><div class="line">    &#125;</div></pre></td></tr></table></figure></p>
<p>在StreamEngine中的in_event()中通过一个read方法将其读到ByteBuffer中，然后解码后放到SessionBase的Pipe中。SocketBase通过他持有的Pipe来获取SessionBase中Pipe的数据，并最终返回给客户端。</p>
<p>对发送与接收部分的小结：</p>
<blockquote>
<p>StreamEngine中的in_event与out_event两个方法，这两个方法是真正读，写数据的地方，抛开一些列数据长度的检查，ByteBuffer的设置填充等，<br>最终的就是调用了SocketChannel的read与write方法。<br>但是经过观察，似乎Poller轮询是如果isWriteable为true，走out_event方法时，只有在StreamEngine发送握手信息时才会发生。而其余，无论是客户端发送或者接收信息，走的都是Poller中的isReadable下的in_event。原因之后进行说明。<br>上面说到客户端的读写在Poller走的都是in_event。<br>读很好理解，另一个端点直接通过SocketChannel,write一个消息过来，因为我们的SocketChannel在Poller中注册了，所以自然就比较能检测到isReadable信号，从而进入in_event<br>但是写的话，由于我们表面上的写操作并不是直接通过SocketChannel来发送消息，而是先将要发送的消息放到Pipe中（其实就是放到一个队列中），然后由之前提过的Signaler发出一个activate_read信号，也就是告诉主机本地有需要发送的消息，请从Pipe中读取。上面也提过，Signaler的消息是通过SinkChannel发送的，对应的SourceChannel在Poller中也注册了。<br>如此，Poller轮询到一个读事件，促发in_event，IOThread从MailBox中读取命令后发现是activate_read命令，触发read_activated事件，该事件由SessionBase开始传播，最终传到StreamEngine中的activate_out事件，由该事件来执行out_event,来真正完成发送数据的过程。</p>
</blockquote>
<hr>
<p>就整个实现过程进行一个小总结</p>
<blockquote>
<p>1.MailBox是核心组件，所有所有动作包括读，写最终都要通过命令传递，才会发生我们预期中的操作<br>2.命令的优先级高于消息，明显的在读写操作前都会先去读MailBox中有没有命令需要执行<br>3.Socket连接方式采用非阻塞模式，具有断线重连，超时重连的功能，重连方式是以指数退步的方式进行<br>4.纵观全局，其实所有的内容都包装在了一个IOObject中，所以层级拆分时还是比较容易的<br>5.整个消息组件用了大量的事件回调，几乎所有的动作都是通过时间回掉来完成的<br>6.整个发送或者或者接收过程，就像是有消息要发送，先发个命令到MailBox，读取命令后执行具体操作。有消息要接收，先发个命令到MailBox，读取命令后执行具体操作。</p>
</blockquote>
<h3 id="5-核心特点"><a href="#5-核心特点" class="headerlink" title="5.  核心特点"></a>5.  核心特点</h3><h5 id="1-嵌入式消息组件"><a href="#1-嵌入式消息组件" class="headerlink" title="1.嵌入式消息组件"></a>1.嵌入式消息组件</h5><p>与rabbitMQ，ActiveMQ有很大的不同，如果说rabbitMQ已经近乎是一个小型操作系统，那么ZeroMQ就像是一个嵌入在操作系统内的一个组件，说白了ZeroMQ就是一组jar包，直接嵌入到项目中就可以运行，它不需要一台独立的服务器来承载整个消息系统。<br>ZeroMQ关注的不是消息的可靠送达，而是着眼于端到端的发送，接收…它希望的是尽快完成任务，而不介意部分消息的丢失。<br>但这也并不是说他完全没有持久化的功能，ZeroMQ是具有一定的本地持久化的功能的，但是能保存的数据量比较有限，而且是暂存于内存中的。</p>
<h5 id="2-高的离谱的吞吐量"><a href="#2-高的离谱的吞吐量" class="headerlink" title="2.高的离谱的吞吐量"></a>2.高的离谱的吞吐量</h5><p>这是网上找到的一张关于MQ的性能分析的图表<br><img src="http://blog.chinaunix.net/attachment/201303/31/22312037_1364725293Ce0Y.png" alt="oxs"><br>显示的是每秒钟发送和接受的消息数。整个过程共产生1百万条1K的消息，测试环境为Windows Vista。从测试数据可以看出，ZeroMQ的性能远远高于其它3个MQ。或者说ZeroMQ与其他3各MQ根本就不再一个量级上比较合适。<br>至于这样的原因跟ZeroMQ的定位，以及对消息的处理方式有很大关联。<br>ZeroMQ对于消息的处理可以说除却请求-应答模式之外，基本就是不关系消息是否丢失，它只管发送。<br>ZeroMQ的定位，它的创始人一直在其社区表示，团队将立志于把ZeroMQ融入到Linux内核中去。<br>基于以上两点，高效的处理速度就成了它必不可少的特点之一。</p>
<h5 id="3-无锁队列与异步模式"><a href="#3-无锁队列与异步模式" class="headerlink" title="3. 无锁队列与异步模式"></a>3. 无锁队列与异步模式</h5><p>之前我们提到了SessionBase与SocketBase之间通过一对Pipe来完成数据交互，在这对Pipe的内部持有的是YPipe的实例，YPipe本质上的实现是一个队列，而且还是一个采用无锁CAS技术的队列。我截出部分代码，供大家参考：<br><figure class="highlight java"><table><tr><td class="gutter"><pre><div class="line">1</div><div class="line">2</div><div class="line">3</div><div class="line">4</div><div class="line">5</div><div class="line">6</div><div class="line">7</div><div class="line">8</div><div class="line">9</div><div class="line">10</div><div class="line">11</div><div class="line">12</div><div class="line">13</div><div class="line">14</div><div class="line">15</div><div class="line">16</div><div class="line">17</div></pre></td><td class="code"><pre><div class="line"><span class="function"><span class="keyword">public</span> <span class="keyword">final</span> <span class="keyword">boolean</span> <span class="title">flush</span> <span class="params">()</span></span></div><div class="line">   &#123;</div><div class="line">       <span class="comment">//  If there are no un-flushed items, do nothing.</span></div><div class="line">       <span class="keyword">if</span> (w == f) &#123;</div><div class="line">           <span class="keyword">return</span> <span class="keyword">true</span>;</div><div class="line">       &#125;</div><div class="line">   </div><div class="line">       <span class="comment">//  Try to set 'c' to 'f'.</span></div><div class="line">       <span class="keyword">if</span> (!c.compareAndSet(w, f)) &#123;</div><div class="line">           c.set (f);</div><div class="line">           w = f;</div><div class="line">           <span class="keyword">return</span> <span class="keyword">false</span>;</div><div class="line">       &#125;</div><div class="line">   </div><div class="line">       w = f;</div><div class="line">       <span class="keyword">return</span> <span class="keyword">true</span>;</div><div class="line">   &#125;</div></pre></td></tr></table></figure></p>
<p>类似于这样的操作在YPipe中还有不少。<br>还是Pipe，我们知道Pipe有一个读端一个写端，这两端均在Poller中注册了异步事件，在该Pipe上发生读写操作时就会触发相应的具体读写事件实现方法。这样的做法提高的程序的响应速度。但是缺点是除非明确在Poller中注册的对象到底是什么，否则比较难判断出将要执行的回调事件将在哪里发生，比如同样是in_event()，IOObject实现了，SocketBase实现了，SessionBase也实现了，如果我们不知道注册时传入的handle是什么，判断起来就有些繁琐了。</p>
<h5 id="4-多核下的线程绑定"><a href="#4-多核下的线程绑定" class="headerlink" title="4.多核下的线程绑定"></a>4.多核下的线程绑定</h5><p>传统的多线程并发模式一般会采用锁，临界区，信号量等技术来控制，而ZeroMQ给出的建议是：在创建IO时不要超出CPU核数。<br>当我们创建一个上下文时都会有这么一句代码“Context context = ZMQ.context(1);”这里就指定了IO线程数。通常来说一个线程足矣。但是如果希望创建多个IO线程，最好不要超出CPU核数，因为此时ZeroMQ会将工作线程其实就是那个Poller绑定到每一个核，免除了线程上下文切换带来的开销。</p>
<h3 id="6-与其他MQ的对比"><a href="#6-与其他MQ的对比" class="headerlink" title="6. 与其他MQ的对比"></a>6. 与其他MQ的对比</h3><p>关于ZeroMQ与其他几个MQ之间的比较，我们在TPS，并发性，持久化，技术点以及扩展性这几个方面进行展开。</p>
<p>#####1.TPS<br>之前提过，这里在说一下。<br><img src="http://blog.chinaunix.net/attachment/201303/31/22312037_1364725293Ce0Y.png" alt="oxs"><br>显示的是每秒钟发送和接受的消息数。整个过程共产生1百万条1K的消息，测试环境为Windows Vista。<br>明显的ZeroMQ最好，其余三者差不多。</p>
<p>#####2.持久化消息比较<br>ZeroMq原生是不支持的，仅支持相当有限的本地缓存，如需要消息持久化需要自己进行扩展。<br>ActiveMq和rabbitMq都支持。</p>
<p>#####3.并发性<br>虽然ZeroMQ在高并发环境下不会出问题，但是有可能会导致本地的缓存区被塞满而导致消息丢失的情况。所以不推荐在并发量较高的情境下使用ZeroMQ.<br>查了下资料发现，ActiveMQ在发送到queue的消息并发较多时，消费端只能接收一部分，比如100条消息在较短的时间内发入，总有10来条接收不到，存放在服务器上，而且这些消息一直不能主动发送出来，后面继续进入的消息都能正常处理，最终只有重新启动服务消费端才能接收到那部分剩下的消息。<br>而RabbitMQ，从实现语言来看，它是并发性最好的，原因是它的实现语言是天生具备高并发高可用的erlang语言。</p>
<p>#####4.技术点以及扩展性<br>ok，首先就扩展性而言，那毫无疑问的是ZeroMQ最强，其余其中MQ都已经是成形的产品，已经是一款应用程序了。而ZeroMQ说白了就是一组库函数。基于这种情况，我们可以按自己的需要实现IPollEvent以及ZObject来开发适合自己的Socket组件，至于它对于消息持久化的不支持，只是原生不支持，因为它的定位不是吃保证可靠的消息传输。所以在可靠性这部分我们完全可以按自己的需求进行扩展。一组lib的扩展度明显是宽于产品级的rabbitMQ之类的产品。<br>技术上虽然ZeroMQ立志于成为Linux内核的消息组件，但是不得不说它的开源社区活跃度是远远不及RabbitMQ或者ActiveMQ。或许是处于它的可靠性考虑，它的应用场景比较受限制。<br>可靠性上虽然ActiveMQ也具备，只是性能上相比于RabbitMQ还是有一定差距，所以大部分的MQ选型都是RabbitMQ。</p>

      
    </div>

    <div>
      
        

      
    </div>

    <div>
      
        

      
    </div>

    <footer class="post-footer">
      
        <div class="post-tags">
          
            <a href="/tags/zeromq/" rel="tag">#zeromq</a>
          
            <a href="/tags/mq/" rel="tag">#mq</a>
          
        </div>
      

      
        <div class="post-nav">
          <div class="post-nav-next post-nav-item">
            
              <a href="/2016/08/28/consul-config/" rel="next" title="consul 配置---K/V存储及ACL">
                <i class="fa fa-chevron-left"></i> consul 配置---K/V存储及ACL
              </a>
            
          </div>

          <div class="post-nav-prev post-nav-item">
            
              <a href="/2016/08/28/spring-ioc-1/" rel="prev" title="源码角度解析SpringIoC的机制">
                源码角度解析SpringIoC的机制 <i class="fa fa-chevron-right"></i>
              </a>
            
          </div>
        </div>
      

      
      
    </footer>
  </article>



    <div class="post-spread">
      
    </div>
  </div>


          </div>
          


          
  <div class="comments" id="comments">
    
      <div class="ds-thread" data-thread-key="2016/08/28/ZeroMQ/"
           data-title="从源码分析ZeroMQ" data-url="http://yoursite.com/2016/08/28/ZeroMQ/">
      </div>
    
  </div>


        </div>
        
          
  
  <div class="sidebar-toggle">
    <div class="sidebar-toggle-line-wrap">
      <span class="sidebar-toggle-line sidebar-toggle-line-first"></span>
      <span class="sidebar-toggle-line sidebar-toggle-line-middle"></span>
      <span class="sidebar-toggle-line sidebar-toggle-line-last"></span>
    </div>
  </div>

  <aside id="sidebar" class="sidebar">
    <div class="sidebar-inner">

      

      
        <ul class="sidebar-nav motion-element">
          <li class="sidebar-nav-toc sidebar-nav-active" data-target="post-toc-wrap" >
            文章目录
          </li>
          <li class="sidebar-nav-overview" data-target="site-overview">
            站点概览
          </li>
        </ul>
      

      <section class="site-overview sidebar-panel ">
        <div class="site-author motion-element" itemprop="author" itemscope itemtype="http://schema.org/Person">
          <img class="site-author-image" itemprop="image"
               src="/uploads/touxiang.jpg"
               alt="钱伟峰" />
          <p class="site-author-name" itemprop="name">钱伟峰</p>
          <p class="site-description motion-element" itemprop="description">每天进步一点点</p>
        </div>
        <nav class="site-state motion-element">
          <div class="site-state-item site-state-posts">
            <a href="/archives">
              <span class="site-state-item-count">21</span>
              <span class="site-state-item-name">日志</span>
            </a>
          </div>

          
            <div class="site-state-item site-state-categories">
              <a href="/categories">
                <span class="site-state-item-count">7</span>
                <span class="site-state-item-name">分类</span>
              </a>
            </div>
          

          
            <div class="site-state-item site-state-tags">
              <a href="/tags">
                <span class="site-state-item-count">11</span>
                <span class="site-state-item-name">标签</span>
              </a>
            </div>
          

        </nav>

        

        <div class="links-of-author motion-element">
          
        </div>

        
        

        
        

      </section>

      
        <section class="post-toc-wrap motion-element sidebar-panel sidebar-panel-active">
          <div class="post-toc">
            
              
            
            
              <div class="post-toc-content"><ol class="nav"><li class="nav-item nav-level-1"><a class="nav-link" href="#ZeroMQ的研究与学习"><span class="nav-number">1.</span> <span class="nav-text">ZeroMQ的研究与学习</span></a><ol class="nav-child"><li class="nav-item nav-level-2"><a class="nav-link" href="#ZeroMQ的一百字概括"><span class="nav-number">1.1.</span> <span class="nav-text">ZeroMQ的一百字概括</span></a><ol class="nav-child"><li class="nav-item nav-level-3"><a class="nav-link" href="#1-简介"><span class="nav-number">1.1.1.</span> <span class="nav-text">1. 简介</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#2-工作模式"><span class="nav-number">1.1.2.</span> <span class="nav-text">2. 工作模式</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#3-层级模型与交互逻辑"><span class="nav-number">1.1.3.</span> <span class="nav-text">3. 层级模型与交互逻辑</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#4-实现原理"><span class="nav-number">1.1.4.</span> <span class="nav-text">4. 实现原理</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#5-核心特点"><span class="nav-number">1.1.5.</span> <span class="nav-text">5.  核心特点</span></a><ol class="nav-child"><li class="nav-item nav-level-5"><a class="nav-link" href="#1-嵌入式消息组件"><span class="nav-number">1.1.5.0.1.</span> <span class="nav-text">1.嵌入式消息组件</span></a></li><li class="nav-item nav-level-5"><a class="nav-link" href="#2-高的离谱的吞吐量"><span class="nav-number">1.1.5.0.2.</span> <span class="nav-text">2.高的离谱的吞吐量</span></a></li><li class="nav-item nav-level-5"><a class="nav-link" href="#3-无锁队列与异步模式"><span class="nav-number">1.1.5.0.3.</span> <span class="nav-text">3. 无锁队列与异步模式</span></a></li><li class="nav-item nav-level-5"><a class="nav-link" href="#4-多核下的线程绑定"><span class="nav-number">1.1.5.0.4.</span> <span class="nav-text">4.多核下的线程绑定</span></a></li></ol></li></ol></li><li class="nav-item nav-level-3"><a class="nav-link" href="#6-与其他MQ的对比"><span class="nav-number">1.1.6.</span> <span class="nav-text">6. 与其他MQ的对比</span></a></li></ol></li></ol></li></ol></div>
            
          </div>
        </section>
      

    </div>
  </aside>


        
      </div>
    </main>

    <footer id="footer" class="footer">
      <div class="footer-inner">
        <div class="copyright" >
  
  &copy; 
  <span itemprop="copyrightYear">2017</span>
  <span class="with-love">
    <i class="fa fa-heart"></i>
  </span>
  <span class="author" itemprop="copyrightHolder">钱伟峰</span>
</div>

<div class="powered-by">
  由 <a class="theme-link" href="https://hexo.io">Hexo</a> 强力驱动
</div>

<div class="theme-info">
  主题 -
  <a class="theme-link" href="https://github.com/iissnan/hexo-theme-next">
    NexT.Muse
  </a>
</div>

        

        
      </div>
    </footer>

    <div class="back-to-top">
      <i class="fa fa-arrow-up"></i>
    </div>
  </div>

  

<script type="text/javascript">
  if (Object.prototype.toString.call(window.Promise) !== '[object Function]') {
    window.Promise = null;
  }
</script>









  



  
  <script type="text/javascript" src="/vendors/jquery/index.js?v=2.1.3"></script>

  
  <script type="text/javascript" src="/vendors/fastclick/lib/fastclick.min.js?v=1.0.6"></script>

  
  <script type="text/javascript" src="/vendors/jquery_lazyload/jquery.lazyload.js?v=1.9.7"></script>

  
  <script type="text/javascript" src="/vendors/velocity/velocity.min.js?v=1.2.1"></script>

  
  <script type="text/javascript" src="/vendors/velocity/velocity.ui.min.js?v=1.2.1"></script>

  
  <script type="text/javascript" src="/vendors/fancybox/source/jquery.fancybox.pack.js?v=2.1.5"></script>


  


  <script type="text/javascript" src="/js/src/utils.js?v=5.0.1"></script>

  <script type="text/javascript" src="/js/src/motion.js?v=5.0.1"></script>



  
  

  
  <script type="text/javascript" src="/js/src/scrollspy.js?v=5.0.1"></script>
<script type="text/javascript" src="/js/src/post-details.js?v=5.0.1"></script>



  


  <script type="text/javascript" src="/js/src/bootstrap.js?v=5.0.1"></script>



  

  
    
  

  <script type="text/javascript">
    var duoshuoQuery = {short_name:"qianweifeng"};
    (function() {
      var ds = document.createElement('script');
      ds.type = 'text/javascript';ds.async = true;
      ds.id = 'duoshuo-script';
      ds.src = (document.location.protocol == 'https:' ? 'https:' : 'http:') + '//static.duoshuo.com/embed.js';
      ds.charset = 'UTF-8';
      (document.getElementsByTagName('head')[0]
      || document.getElementsByTagName('body')[0]).appendChild(ds);
    })();
  </script>

  
    
    <script src="/vendors/ua-parser-js/dist/ua-parser.min.js?v=0.7.9"></script>
    <script src="/js/src/hook-duoshuo.js"></script>
  






  
  

  

  

  

</body>
</html>
